hermes-skills/skills/devops/python-web-service-deployment/references/fastapi-with-auth-draft-scoring.md

FastAPI Backend Patterns — Auth, Leagues, Snake Draft, Scoring

Built Jul 8, 2026 for the Shark Attack Fantasy Game. Monolithic server.py (no routers/blueprints), SQLite, bcrypt+JWT, league system with snake draft, and event-based scoring with auto-recalculation.

Architecture Pattern

A single-file FastAPI app (server.py) containing everything:

server.py          ← All models, routes, DB setup, auth, helpers
seed.py            ← Standalone region/schema seeder (optional, built-in init also works)
requirements.txt   ← fastapi, uvicorn, bcrypt, pyjwt
venv/              ← Python virtualenv
game.db            ← SQLite database (created on first startup)

Rationale for single-file: For services with < 20 endpoints, a monolithic file avoids import overhead, is easier to scroll/debug, and can be split later when it crosses ~800 lines.

Database Pattern (SQLite + sqlite3)

Raw sqlite3 — no ORM. Schema created via executescript() at startup:

def get_db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    conn.row_factory = sqlite3.Row
    return conn

Key choices:

• sqlite3.Row row factory — dict-compatible access, dict(row) for serialization
• WAL mode for concurrent reads
• Foreign keys enforced at DB level
• executescript() for multi-statement DDL at init
• New get_db() connection per request (cheap with SQLite + WAL)
• Always conn.close() after each request in sync handlers

Auth Pattern (bcrypt + PyJWT)

import bcrypt, jwt

def hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()

def verify_password(password: str, password_hash: str) -> bool:
    return bcrypt.checkpw(password.encode(), password_hash.encode())

def create_token(user_id: int) -> str:
    payload = {
        "user_id": user_id,
        "exp": datetime.utcnow() + timedelta(days=30),
        "iat": datetime.utcnow(),
    }
    return jwt.encode(payload, JWT_SECRET, algorithm="HS256")

async def get_current_user(authorization: Optional[str] = Header(None)) -> dict:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing or invalid authorization header")
    token = authorization.split(" ", 1)[1]
    payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
    user = get_db().execute("SELECT id, email, display_name FROM users WHERE id = ?", (payload["user_id"],)).fetchone()
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return dict(user)

Pattern notes:

• Depends(get_current_user) on any protected route — returns a dict of db row
• Token expiry set at 30 days, tracked as exp claim
• JWT_SECRET should come from env var; dev fallback is fine
• No session refresh endpoint (can add later if needed)

League + Membership Pattern (join tables)

Three tables for the league system:

leagues (id, name, invite_code, status, season_start, season_end, created_by, created_at)
league_members (id, league_id, user_id, draft_order, is_commissioner)
draft_picks (id, league_id, user_id, region_id, round, pick_number)

Invite code generation:

import secrets
def generate_invite_code() -> str:
    return secrets.token_hex(4).upper()  # 8-char hex, retry on collision

Commissioner check pattern:

def get_commissioner_membership(conn, league_id, user_id):
    member = conn.execute(
        "SELECT * FROM league_members WHERE league_id = ? AND user_id = ? AND is_commissioner = 1",
        (league_id, user_id),
    ).fetchone()
    if not member:
        raise HTTPException(status_code=403, detail="Only the commissioner can perform this action")

Snake Draft Logic

def get_current_round_and_pick(conn, league_id, member_count):
    picks_made = conn.execute(
        "SELECT COUNT(*) FROM draft_picks WHERE league_id = ?", (league_id,)
    ).fetchone()[0]

    if picks_made >= member_count * 12:  # max rounds
        return 13, None, True  # draft complete

    current_round = (picks_made // member_count) + 1
    pick_in_round = picks_made % member_count

    members = conn.execute(
        """SELECT lm.user_id, u.display_name
           FROM league_members lm JOIN users u ON u.id = lm.user_id
           WHERE lm.league_id = ?
           ORDER BY lm.draft_order ASC""",
        (league_id,),
    ).fetchall()

    if current_round % 2 == 1:
        next_user = members[pick_in_round]   # forward order (odd rounds)
    else:
        next_user = members[member_count - 1 - pick_in_round]  # reverse (even rounds)

    return current_round, dict(next_user) if next_user else None, False

Snake reversal: Odd rounds = natural draft_order ASC, even rounds = iterate backwards. Member count must be known to compute the reverse index.

Turn enforcement: Before allowing a pick, compare next_pick["user_id"] against the requesting user's ID. Return 403 if mismatch.

Duplicate protection: UNIQUE(league_id, region_id) constraint on draft_picks table, with a 409 check before insertion.

Draft status flow: draft → active → closed. Commissioner-only start transition. Draft start requires ≥2 members.

Score Event System

Events table plus materialized user scores:

scores (id, region_id, event_type, description, source_url, points, event_date, verified, created_at)
user_scores (user_id, league_id, total_points)  -- PRIMARY KEY (user_id, league_id)

Point values: sighting=2 (changed from 1), bite=5, fatality=10

Auto-recalculation: Every time a score event is ingested, recalculate for all leagues that drafted that region:

def recalculate_user_scores(conn, league_id):
    conn.execute("DELETE FROM user_scores WHERE league_id = ?", (league_id,))
    users = conn.execute("SELECT DISTINCT user_id FROM draft_picks WHERE league_id = ?", (league_id,)).fetchall()
    for user in users:
        total = conn.execute(
            """SELECT COALESCE(SUM(s.points), 0)
               FROM scores s JOIN draft_picks dp ON s.region_id = dp.region_id
               WHERE dp.league_id = ? AND dp.user_id = ?""",
            (league_id, user["user_id"]),
        ).fetchone()[0]
        conn.execute("INSERT INTO user_scores (...) VALUES (?, ?, ?)", (user_id, league_id, total))

Important: Recalculation happens on every score ingestion for every affected league. With small datasets (dozens of users, hundreds of events) this is fine — the full table DELETE + INSERT loop is sub-100ms. For larger scale, switch to incremental updates or a background task.

Verification / Testing Pattern

Standalone test scripts (no pytest) are preferred for fast iteration:

def api(method, path, data=None, token=None):
    url = f"{BASE}{path}"
    body = json.dumps(data).encode() if data else None
    req = urllib.request.Request(url, data=body, method=method)
    req.add_header("Content-Type", "application/json")
    if token: req.add_header("Authorization", f"Bearer {token}")
    try:
        with urllib.request.urlopen(req) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        return {"error": e.code, "detail": json.loads(e.read())}

This avoids pytest dependency and works with any Python. The error / detail pattern in the return value mirrors the FastAPI error shape and lets checks be uniform: check(result.get("error") == 403, "non-commissioner blocked").

Startup Pattern

@app.on_event("startup")
def on_startup():
    init_db()
    seed_regions()

Deprecated in newer FastAPI but works fine. Use lifespan context manager for new projects:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    seed_regions()
    yield

app = FastAPI(lifespan=lifespan)

Pitfalls

• sqlite3 doesn't autocommit DML — always call conn.commit() after INSERT/UPDATE/DELETE.
• PRAGMA foreign_keys=ON must be set per connection, not just at schema creation.
• JWT secret in code is a dev pattern — always override via JWT_SECRET env var in production.
• Auto-recalculation costs scale with number of leagues × drafted regions — for 100+ simultaneous leagues, move to an event queue + batch recalculation.
• FastAPI sync vs async — all handlers above use sync DB calls. FastAPI runs sync def handlers in a thread pool, which is fine for SQLite. Use async def only when the endpoint does IO (HTTP calls, SSE streaming).
• sqlite3.Row objects are not JSON-serializable — call dict(row) before returning.
• Pydantic v1 vs v2 — modern FastAPI uses Pydantic v2. Use BaseModel from pydantic (not pydantic.v1). String Field(min_length=...) works in both.