Rolf
3a1da1482c
- FastAPI + SQLite + Alpine.js + HTMX PWA for Bingo Bango Bongo golf scoring - Passwordless magic link auth with email (aiosmtplib) and admin invite system - Real-time score entry via WebSocket with drag-and-drop player ordering - Course management with per-hole par and stroke index - Scorecard with golf score markers (birdie, eagle, bogey, etc.) - Two-step new game form with tee selection per player - Dashboard with stats, live games social stream, and recent game history - Game summary view (read-only scorecard with leaderboard) - User profile pages with win stats - PWA install support with generated PNG icons - Admin panel for users, courses, games, and invitations Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
94 lines
2.5 KiB
Python
94 lines
2.5 KiB
Python
import os
|
|
import secrets
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import aiosqlite
|
|
|
|
from app.email import send_magic_link
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SITE_URL = os.getenv("SITE_URL", "http://localhost:8000")
|
|
ENV = os.getenv("ENV", "development")
|
|
TOKEN_TTL_MINUTES = 15
|
|
SESSION_TTL_DAYS = 30
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _in(minutes: int = 0, days: int = 0) -> str:
|
|
delta = timedelta(minutes=minutes, days=days)
|
|
return (datetime.now(timezone.utc) + delta).isoformat()
|
|
|
|
|
|
def _expired(iso: str) -> bool:
|
|
return datetime.now(timezone.utc) > datetime.fromisoformat(iso)
|
|
|
|
|
|
async def create_magic_token(email: str, db: aiosqlite.Connection) -> str:
|
|
# Clean up old unused tokens for this email
|
|
await db.execute(
|
|
"DELETE FROM magic_link_tokens WHERE email = ? AND used_at IS NULL",
|
|
(email,),
|
|
)
|
|
|
|
token_id = secrets.token_urlsafe(16)
|
|
token = secrets.token_urlsafe(32)
|
|
expires_at = _in(minutes=TOKEN_TTL_MINUTES)
|
|
|
|
await db.execute(
|
|
"INSERT INTO magic_link_tokens (id, email, token, expires_at) VALUES (?, ?, ?, ?)",
|
|
(token_id, email, token, expires_at),
|
|
)
|
|
await db.commit()
|
|
|
|
link = f"{SITE_URL}/auth/verify?token={token}"
|
|
await send_magic_link(email, link)
|
|
return token
|
|
|
|
|
|
async def verify_magic_token(token: str, db: aiosqlite.Connection):
|
|
"""Returns the email if token is valid and unused, else None."""
|
|
async with db.execute(
|
|
"SELECT email, expires_at, used_at FROM magic_link_tokens WHERE token = ?",
|
|
(token,),
|
|
) as cursor:
|
|
row = await cursor.fetchone()
|
|
|
|
if row is None:
|
|
return None
|
|
if row["used_at"] is not None:
|
|
return None
|
|
if _expired(row["expires_at"]):
|
|
return None
|
|
|
|
# Mark token as used
|
|
await db.execute(
|
|
"UPDATE magic_link_tokens SET used_at = ? WHERE token = ?",
|
|
(_now(), token),
|
|
)
|
|
await db.commit()
|
|
|
|
return row["email"]
|
|
|
|
|
|
async def create_session(user_id: str, db: aiosqlite.Connection) -> str:
|
|
session_id = secrets.token_urlsafe(16)
|
|
token = secrets.token_urlsafe(32)
|
|
expires_at = _in(days=SESSION_TTL_DAYS)
|
|
|
|
await db.execute(
|
|
"INSERT INTO sessions (id, user_id, token, expires_at, created_at) VALUES (?, ?, ?, ?, ?)",
|
|
(session_id, user_id, token, expires_at, _now()),
|
|
)
|
|
await db.commit()
|
|
return token
|
|
|
|
|
|
async def delete_session(token: str, db: aiosqlite.Connection):
|
|
await db.execute("DELETE FROM sessions WHERE token = ?", (token,))
|
|
await db.commit()
|