1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
|
from pathlib import Path
import json
import hmac
from typing import Annotated
import secrets
import random
from enum import Enum
from dataclasses import dataclass
import datetime
from contextlib import asynccontextmanager
from typing import Any
from collections.abc import AsyncGenerator, Sequence
from sqlalchemy import select, String
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.dialects.postgresql import ARRAY
from litestar import Litestar, get, post, Request
from litestar.contrib.jinja import JinjaTemplateEngine
from litestar.exceptions import HTTPException
from litestar.template.config import TemplateConfig
from litestar.response import Template, Redirect
from litestar.static_files import create_static_files_router
from litestar.enums import RequestEncodingType
from litestar.params import Body
from litestar import Request
from litestar.datastructures import State
# automake calendar invite ics
# use url_fors, timezones?
class Base(DeclarativeBase):
pass
class Event(Base):
__tablename__ = "events"
def __init__(self, iden, password, title, time, location, description):
self.iden = iden
self.password = password
self.title = title
self.time = time
self.location = location
self.description = description
self.invites = '[]'
iden: Mapped[str] = mapped_column(primary_key=True)
password: Mapped[str]
title: Mapped[str]
time: Mapped[datetime.datetime]
location: Mapped[str]
description: Mapped[str]
invites: Mapped[str]
def get_invites(self):
return json.loads(self.invites)
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///symposium.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
parts = [
"Oxbow", "Limber", "Steadfast", "Regicide", "Swarm", "Weave", "Bough", "Canopy", "Herald", "Scorn",
"Alder", "Aerial", "Welkin", "Acrid", "Kindling", "Rapture", "Myrtle", "Envy", "Solstice",
"Juniper", "Cleaving", "Stream", "Reaper", "Sluice", "Conduit", "Disdain", "Sylvan", "Ravish", "Atrium",
"Thresh", "Harvest", "Water", "Renewal", "Rosy", "Frieze", "Portal", "Vespers", "Litany", "Serpent",
"Primate", "Incite", "Canon", "Acquiese", "Mirror", "Script", "Seal", "Privy",
"Piercing", "Heresy", "Subduct", "Sceptre", "Arrogance", "Ivory", "Accrete", "Cluster",
"Sepulchre", "Summon", "Pleading", "Myriad", "Exalted", "Sentry", "Shriven", "River", "Threshold"
]
def gen_iden():
return "-".join(random.choices(parts, k=8))
iden = 'Weave-Stream-Limber-Exalted-Sluice-Reaper-Myrtle-Incite'
title = 'Hot Pot at the Cafe Cyfrae Violae'
when = datetime.datetime.fromisoformat("2024-04-07T13:00:00")
where = '133 E 4th St, Apt 6, New York NY 10003'
what = "Hot pot, Anxi oolong, baijiu, Hua Zhou."
EVENTS = {}
@get("/")
async def index() -> Template:
return Template(template_name="index.html")
@get("/event/{iden:str}")
async def event(state: State, iden: str, password: str = "") -> Template:
async with sessionmaker(bind=state.engine) as session:
async with session.begin():
query = select(Event).where(Event.iden == iden)
result = await session.execute(query)
event = result.scalar_one()
manage = False
if password and hmac.compare_digest(event.password, password):
manage = True
context = dict(event=event, manage=manage)
return Template(template_name="event.html", context=context)
@dataclass
class EditRequest:
title: str
when: str
where: str
what: str
@post("/event/create")
async def create(state: State, data: Annotated[EditRequest, Body(media_type=RequestEncodingType.URL_ENCODED)]) -> Redirect:
iden = gen_iden()
password = secrets.token_bytes(16).hex()
event = Event(iden, password, data.title, datetime.datetime.fromisoformat(data.when), data.where, data.what)
async with sessionmaker(bind=state.engine) as session:
async with session.begin():
session.add(event)
return Redirect(path="/symposium/event/" + iden + "?password=" + event.password)
@post("/event/{iden:str}/edit")
async def edit(state: State, request: Request, iden: str, password: str, data: Annotated[EditRequest, Body(media_type=RequestEncodingType.URL_ENCODED)]) -> Redirect: #-> Template:
async with sessionmaker(bind=state.engine) as session:
async with session.begin():
query = select(Event).where(Event.iden == iden)
result = await session.execute(query)
event = result.scalar_one()
manage = False
if password and hmac.compare_digest(event.password, password):
manage = True
if not manage:
raise ValueError("no auth")
event.title = data.title
event.time = datetime.datetime.fromisoformat(data.when)
event.location = data.where
event.description = data.what
return Redirect(path="/symposium/event/" + iden + "?password=" + event.password)
@dataclass
class JoinRequest:
name: str
@post("/event/{iden:str}/join")
async def join(state: State, request: Request, iden: str, data: Annotated[JoinRequest, Body(media_type=RequestEncodingType.URL_ENCODED)], password: str = "") -> Redirect: #-> Template:
async with sessionmaker(bind=state.engine) as session:
async with session.begin():
query = select(Event).where(Event.iden == iden)
result = await session.execute(query)
event = result.scalar_one()
name = data.name
# TODO if name exists reject
invites = json.loads(event.invites)
invites.append(name)
event.invites = json.dumps(invites)
url = "/symposium/event/" + iden
if password:
url += "?password=" + password
return Redirect(path=url)
app = Litestar(
route_handlers=[
index,
event,
create,
edit,
join,
create_static_files_router(path='/static', directories=['static']),
],
template_config=TemplateConfig(
directory=Path("templates"),
engine=JinjaTemplateEngine,
),
lifespan=[db_connection],
)
|