# 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:

```python
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)

```python
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:

```sql
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:**
```python
import secrets
def generate_invite_code() -> str:
    return secrets.token_hex(4).upper()  # 8-char hex, retry on collision
```

**Commissioner check pattern:**
```python
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

```python
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:

```sql
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:

```python
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:

```python
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

```python
@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:

```python
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.