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
|
from pathlib import Path
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
# link location to gmaps
# automake calendar invite ics
# persistence
# 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]
@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(request: Request, iden: str, password: str, data: Annotated[EditRequest, Body(media_type=RequestEncodingType.URL_ENCODED)]) -> Redirect: #-> Template:
event = EVENTS[iden]
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 = 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(request: Request, iden: str, data: Annotated[JoinRequest, Body(media_type=RequestEncodingType.URL_ENCODED)]) -> Redirect: #-> Template:
event = EVENTS[iden]
name = data.name
# TODO if name exists reject
invites = json.loads(event.invites)
invites.append(name)
event.invites = json.dumps(invites)
return Redirect(path="/symposium/event/" + iden)
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],
debug=True,
)
|
