refactor(api): move from Postgres+RLS to single-engine SQLite
CI / Repo hygiene (pull_request) Successful in 2s
CI / Web (lint, typecheck, build) (pull_request) Successful in 13s
CI / Migrations reversible (pull_request) Successful in 6s
CI / API (lint, types, tests) (pull_request) Successful in 52s

Reverses a shipped, tested, merged decision (D4/PR #2) rather than building
on it — see docs/DECISIONS.md D15 for the full record: what was rejected
(Postgres as a second container; Postgres+PostGIS bundled inside the single
container via a supervisor), what this costs (no database-level RLS, no
PostGIS, procrastinate needs replacing — all stated as a concern before this
was decided, and reaffirmed anyway, which is the user's call to make about
their own instance).

The one invariant-critical consequence: isolation between users now rests
entirely on the repository-layer scope (db.py's `Scope.select()`), not two
layers. CLAUDE.md's invariant #4 is revised accordingly. This is not a
downgrade-and-hope — `Scope` is built so an unfiltered query against a
user-owned table is structurally harder to write than a scoped one (there is
no method on `Scope` that returns one), and
tests/test_auth.py::test_scoped_session_blocks_cross_user_reads replaces the
old RLS proof with the same empirical standard: it doesn't trust the query
builder filters correctly because the code reads correctly, it registers two
real users and checks.
test_unscoped_session_can_see_every_user_when_misused is the deliberately
alarming companion — it demonstrates exactly what a reviewer must now catch,
since nothing else will.

Six real, non-obvious SQLite behaviours found and fixed by actually running
this against a real file, not assumed from docs:

- Foreign keys, ON DELETE CASCADE included, are OFF by default per
  connection — deleting a user silently left orphaned sessions/api_tokens,
  no error either way. Fixed with PRAGMA foreign_keys=ON on every connect.
- Transactions default to DEFERRED, which only takes a write lock on the
  first actual write — a real check-then-act race for invite redemption
  (two concurrent redemptions could both read used_count < max_uses as true
  before either commits). Fixed by disabling the driver's implicit BEGIN and
  issuing BEGIN IMMEDIATE ourselves — SQLAlchemy's own documented recipe for
  this, not improvised.
- DateTime(timezone=True) does NOT round-trip tzinfo on SQLite — a tz-aware
  datetime goes in, a naive one comes back out, and every
  `expires_at < datetime.now(UTC)` comparison in auth/service.py then raises
  TypeError. Fixed once at the Base level with a UTCDateTime TypeDecorator
  rather than per-column.
- Uuid(as_uuid=True) stores as 32-char hex with NO hyphens on SQLite, not
  str(uuid)'s hyphenated form. A test fixture that raw-inserted the
  hyphenated form left rows the ORM's own later UPDATE (via
  invite.used_count += 1's autoflush) could never match by primary key,
  updating zero rows and raising StaleDataError. Fixed by using .hex to
  match exactly what the ORM itself writes.
- BEGIN IMMEDIATE applies to every transaction, reads included — a
  long-lived test fixture that autobegins a transaction via a bare read and
  never explicitly closes it holds SQLite's exclusive write lock for the
  rest of the test, and a later scoped_session() call fails with "database
  is locked". Not an app-code bug (every real session block closes cleanly
  on exit), but real enough to document since the next person writing a
  test against the db_auth fixture will hit it too.
- Python's sqlite3 module deprecates its own implicit datetime adapter as of
  3.12 — silent today, warns on every raw-SQL datetime bind. Only ever hit
  test fixture code (the ORM path never uses it, confirmed by running the
  ORM-only health test with warnings promoted to errors and it stayed
  clean); fixed there with an explicit .isoformat() rather than left for a
  future Python version to turn into a real failure.

Also, since with_for_update() silently no-ops on SQLite (confirmed —
SQLAlchemy emits no SQL for it, no error either) rather than actually
locking anything: removed it from register()'s invite-redemption query and
corrected the comment to attribute the concurrency guarantee to BEGIN
IMMEDIATE, where it now actually lives.

One PR, not several, for the same reason PR #2 was: the migration, the
models, db.py, and the docs recording why are five views of one decision —
splitting them wouldn't make review easier, just disconnected. 552
insertions / 548 deletions across 17 files, most of it necessarily touching
what PR #2 shipped rather than net-new code.

Deliberately deferred, not solved here: PostGIS's replacement for spatial
storage, procrastinate's replacement for background jobs, and the
EXCLUDE USING gist constraint's replacement for component_installs — none
of those tables exist yet (Phase 1-2), so none of it is broken, and
docs/DECISIONS.md D15 records exactly what each future phase needs to
decide before it can be built. .gitea/workflows/deploy pipeline (PR #4,
built for the old 3-container Postgres compose stack) was closed as
superseded rather than merged; the single-container image build is
follow-up work, not part of this change.

Verified: ruff check, ruff format --check, and mypy --strict all clean.
13/13 pytest passing against a real SQLite file, including with
DeprecationWarning promoted to an error (confirms the sqlite3 adapter
deprecation fix actually holds, not just that it's quiet by default). Full
alembic upgrade -> downgrade -1 -> upgrade cycle run clean. alembic check
clean with no include_object filter needed at all now (SQLite starts with
nothing but what our own migrations create — no PostGIS/TIGER noise to
filter out in the first place). CI's exact migration command sequence
reproduced locally end to end before touching the workflow file.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-21 15:42:44 -04:00
co-authored by Claude Opus 5
parent 1832059e03
commit e7392a5723
17 changed files with 548 additions and 544 deletions
+2 -2
View File
@@ -10,8 +10,8 @@ router = APIRouter(tags=["health"])
async def healthz() -> dict[str, str]:
"""Liveness + a real database round trip.
Not RLS-scoped (there's no user yet at this point) — see db.unscoped_session's docstring for
why that's safe: it can see zero rows of any user-owned table regardless.
Uses `unscoped_session()`, not `scoped_session()` — there's no user yet at this point, and
`SELECT 1` never touches a user-owned table anyway. See db.py's module docstring.
"""
async with unscoped_session() as db:
await db.execute(text("SELECT 1"))
+12 -10
View File
@@ -1,9 +1,8 @@
"""Auth bootstrap logic: register, login, session validation, logout.
Every function here runs against the BYPASSRLS `auth` database role (see db.py's module
docstring for why) and every query is an exact match on a unique key — email, token_hash, or
code_hash — never an unfiltered scan. That's what makes bypassing RLS safe here: there's no
"list everything" code path for these functions to accidentally expose.
Every function here runs against `db.auth_session()` (see its docstring for why these specific
pre-identity lookups need it) and every query is an exact match on a unique key — email,
token_hash, or code_hash — never an unfiltered scan.
Nothing outside this module should import `db.auth_session` — if a new feature needs it, that's
a sign the feature belongs here, not that the import should spread.
@@ -59,17 +58,20 @@ async def register(
) -> AuthenticatedSession:
"""Validate an invite and create a user, atomically.
Open signup does not exist — see docs/PLAN.md "Auth". The `SELECT ... FOR UPDATE` on the
invite row is what stops a shared invite link being redeemed twice concurrently; without it,
two requests could both read `used_count < max_uses` as true before either commits.
Open signup does not exist — see docs/PLAN.md "Auth". What stops a shared invite link being
redeemed twice concurrently is db.py's `BEGIN IMMEDIATE` setup, not a row lock on this
SELECT — SQLite has no `SELECT ... FOR UPDATE` (SQLAlchemy's SQLite dialect silently no-ops
`.with_for_update()`, confirmed empirically; it used to appear here when this ran against
Postgres — see git history). `BEGIN IMMEDIATE` takes SQLite's write lock for the whole
transaction up front, so two concurrent redemptions can't both read `used_count < max_uses`
as true before either commits — the second one simply waits for the first transaction to
finish, then sees the incremented count.
"""
code_hash = hash_invite_code(invite_code)
async with auth_session() as db:
async with db.begin():
invite = (
await db.execute(
select(Invite).where(Invite.code_hash == code_hash).with_for_update()
)
await db.execute(select(Invite).where(Invite.code_hash == code_hash))
).scalar_one_or_none()
if invite is None:
+8 -9
View File
@@ -1,24 +1,23 @@
"""Application settings, read from environment variables.
Two separate database DSNs are deliberate, not an oversight — see db.py for why: one connects
as a role with BYPASSRLS (used only by the auth bootstrap path, which must look identity up
*before* it can be scoped), the other as a normal RLS-subject role (used for every other query).
Single SQLite database (see docs/DECISIONS.md D15 for why this isn't the two-role Postgres+RLS
setup Phase 0 originally shipped with) — one DSN, one engine, no BYPASSRLS/NOBYPASSRLS split.
Isolation between users is enforced entirely by the repository-layer scope now; see db.py and
CLAUDE.md's invariant #4.
"""
from functools import lru_cache
from pydantic import PostgresDsn
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="VELODROME_", extra="ignore")
# Scoped role: NOBYPASSRLS. Every request-handling query outside the auth bootstrap uses this.
database_url_app: PostgresDsn
# Bootstrap role: BYPASSRLS. Used ONLY by velodrome.auth.service for the pre-identity lookups
# (login by email, session-by-token, invite-by-code) and for creating new users/sessions.
database_url_auth: PostgresDsn
# A SQLAlchemy URL, e.g. sqlite+aiosqlite:////data/velodrome.db. Not typed as a stricter DSN
# (pydantic has no built-in sqlite+aiosqlite validator) — Alembic and the app both read this
# same setting, so keep it a plain string rather than inventing a validator neither needs.
database_url: str = "sqlite+aiosqlite:///./velodrome.db"
# AES-GCM key for encrypting third-party credentials (e.g. the Bryton digest, added in a later
# phase). Not used yet in Phase 0, but declared now so the settings shape is stable.
+107 -77
View File
@@ -1,113 +1,143 @@
"""Two database engines, on purpose.
"""Single-engine SQLite access.
`app` engine: connects as a role with RLS enforced (NOBYPASSRLS). Every request handler that has
already established who the caller is uses this, wrapped in `scoped_session()` below, which sets
`app.user_id` for the transaction so RLS policies can key off it.
Phase 0 originally ran two Postgres roles with row-level security as a database-enforced
isolation layer (`velodrome_app`/`velodrome_auth` — see docs/DECISIONS.md D4). D15 moved the
database to SQLite, which has no roles, no session variables, and no policy engine — there is no
database-enforced layer left. Isolation between users now rests entirely on the query builder
below. CLAUDE.md's invariant #4 treats this file as load-bearing, not a convenience wrapper: a
new user-owned table without a passing isolation test (see tests/test_auth.py's pattern) is not
done.
`auth` engine: connects as a role with BYPASSRLS. Used ONLY by velodrome.auth.service, and only
for the narrow set of queries that must run *before* identity is known — looking a user up by
email at login, a session up by its hashed token, an invite up by its hashed code — plus the
inserts that create those rows in the first place. Nothing outside auth/service.py should import
this engine; if you find yourself reaching for it elsewhere, the query almost certainly belongs
in a repository method on the scoped session instead (see CLAUDE.md invariant #4).
Two access patterns:
- `scoped_session(user_id)` — for every query against a user-owned table once identity is known.
Yields a `Scope`, whose `select()` is the ONLY way to build a query through it — every query it
builds is pre-filtered to that user_id, on any model that declares a `user_id` column. There is
no method on `Scope` that returns an unfiltered query. This is what makes "forgot the WHERE
clause" structurally harder than remembering to write one by hand.
- `auth_session()` — for velodrome.auth.service ONLY: the handful of pre-identity lookups (login
by email, a session by its token hash, an invite by its code hash) that by definition can't be
scoped to a user_id nobody has established yet. Every query on this session must be an exact
match on a unique key, never an unfiltered scan — that discipline is what made bypassing RLS
safe before, and it's what keeps this safe now that RLS is gone. Nothing outside auth/service.py
should import this.
"""
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any
from uuid import UUID
from sqlalchemy import text
from sqlalchemy import Select, event, select
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.pool import NullPool
from velodrome.config import get_settings
def _make_engine(url: str) -> AsyncEngine:
settings = get_settings()
# NullPool in dev/test keeps behaviour predictable across the pytest-asyncio event loop;
# production tuning (pool_size etc.) is a deploy-time concern, not a Phase-0 one.
return create_async_engine(
url,
poolclass=NullPool if settings.environment == "test" else None,
echo=False,
)
def _configure_sqlite_for_concurrent_writers(eng: AsyncEngine) -> None:
"""Two SQLite defaults that silently do the wrong thing if left alone — confirmed empirically
against a real aiosqlite connection, not assumed from docs:
1. Foreign key enforcement, `ON DELETE CASCADE` included, is OFF by default per connection.
Without `PRAGMA foreign_keys=ON`, deleting a user leaves its sessions/api_tokens rows
behind instead of cascading — verified this happens silently, no error either way.
2. pysqlite/aiosqlite default to a DEFERRED transaction, which only takes a write lock on the
first actual write statement — leaving a real check-then-act race window. Concretely: two
concurrent redemptions of the same invite code could both read `used_count < max_uses` as
true before either commits, double-spending a single-use invite. Disable the driver's own
implicit BEGIN handling and issue `BEGIN IMMEDIATE` ourselves instead, which takes the
write lock at transaction start and correctly serializes writers — SQLAlchemy's own
documented recipe for this, not a workaround improvised here.
"""
@event.listens_for(eng.sync_engine, "connect")
def _on_connect(dbapi_connection: Any, connection_record: Any) -> None:
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
dbapi_connection.isolation_level = None
@event.listens_for(eng.sync_engine, "begin")
def _begin_immediate(conn: Any) -> None:
conn.exec_driver_sql("BEGIN IMMEDIATE")
_app_engine: AsyncEngine | None = None
_auth_engine: AsyncEngine | None = None
_engine: AsyncEngine | None = None
_sessionmaker: async_sessionmaker[AsyncSession] | None = None
def app_engine() -> AsyncEngine:
global _app_engine
if _app_engine is None:
_app_engine = _make_engine(str(get_settings().database_url_app))
return _app_engine
def _engine_instance() -> AsyncEngine:
global _engine
if _engine is None:
settings = get_settings()
url = settings.database_url
# NullPool in dev/test keeps behaviour predictable across the pytest-asyncio event loop;
# production pool tuning is a deploy-time concern, not a Phase-0 one.
_engine = create_async_engine(
url,
poolclass=NullPool if settings.environment == "test" else None,
echo=False,
)
if url.startswith("sqlite"):
_configure_sqlite_for_concurrent_writers(_engine)
return _engine
def auth_engine() -> AsyncEngine:
global _auth_engine
if _auth_engine is None:
_auth_engine = _make_engine(str(get_settings().database_url_auth))
return _auth_engine
_app_sessionmaker: async_sessionmaker[AsyncSession] | None = None
_auth_sessionmaker: async_sessionmaker[AsyncSession] | None = None
def _app_sessions() -> async_sessionmaker[AsyncSession]:
global _app_sessionmaker
if _app_sessionmaker is None:
_app_sessionmaker = async_sessionmaker(app_engine(), expire_on_commit=False)
return _app_sessionmaker
def _auth_sessions() -> async_sessionmaker[AsyncSession]:
global _auth_sessionmaker
if _auth_sessionmaker is None:
_auth_sessionmaker = async_sessionmaker(auth_engine(), expire_on_commit=False)
return _auth_sessionmaker
def _sessions() -> async_sessionmaker[AsyncSession]:
global _sessionmaker
if _sessionmaker is None:
_sessionmaker = async_sessionmaker(_engine_instance(), expire_on_commit=False)
return _sessionmaker
@asynccontextmanager
async def auth_session() -> AsyncIterator[AsyncSession]:
"""A BYPASSRLS session. See module docstring — auth/service.py only."""
async with _auth_sessions()() as session:
"""See module docstring — auth/service.py only."""
async with _sessions()() as session:
yield session
@asynccontextmanager
async def scoped_session(user_id: UUID) -> AsyncIterator[AsyncSession]:
"""An RLS-scoped session for a known, authenticated user.
`SET LOCAL` binds to the current transaction, not the connection, so this is safe under
connection pooling — it can never leak `app.user_id` from one request into a pooled
connection reused by a different request.
"""
async with _app_sessions()() as session:
async with session.begin():
await session.execute(
# bound parameter, not string interpolation — user_id is a UUID we generated
# or validated ourselves, but there is no reason to ever risk it.
text("SELECT set_config('app.user_id', :uid, true)"),
{"uid": str(user_id)},
)
yield session
@asynccontextmanager
async def unscoped_session() -> AsyncIterator[AsyncSession]:
"""An `app`-role session with no `app.user_id` set.
RLS policies default-deny when `current_setting('app.user_id', true)` is NULL, so this sees
zero rows of any user-owned table — useful for health checks and anything that only touches
non-RLS tables. Prefer `scoped_session` whenever a user is known.
"""A plain session with no user scoping applied at all — for health checks and anything that
never touches a user-owned table. Prefer `scoped_session` whenever a user is known; reaching
for this instead of that for a query against a user-owned table is exactly the mistake
invariant #4 exists to catch in review.
"""
async with _app_sessions()() as session:
async with _sessions()() as session:
yield session
class Scope:
"""A user-scoped query builder. `select()` is the only way to build a query through this
object, and it is always pre-filtered to `user_id` — there is no method here that returns an
unfiltered query against a user-owned table.
"""
def __init__(self, session: AsyncSession, user_id: UUID) -> None:
self.session = session
self.user_id = user_id
def select(self, model: type[DeclarativeBase]) -> "Select[Any]":
if not hasattr(model, "user_id"):
raise TypeError(
f"{model.__name__} has no user_id column — it isn't a user-owned table, so "
"scoped_session() is the wrong tool here. Use auth_session() (pre-identity "
"lookups only) or unscoped_session() (health checks etc.) instead."
)
return select(model).where(model.user_id == self.user_id)
@asynccontextmanager
async def scoped_session(user_id: UUID) -> AsyncIterator[Scope]:
"""A `Scope` for a known, authenticated user. See module docstring."""
async with _sessions()() as session:
async with session.begin():
yield Scope(session, user_id)
+34 -7
View File
@@ -1,15 +1,42 @@
from datetime import datetime
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import DateTime
from sqlalchemy.engine import Dialect
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.types import TypeDecorator
class UTCDateTime(TypeDecorator[datetime]):
"""`DateTime(timezone=True)` does NOT round-trip tzinfo on SQLite — confirmed empirically:
a tz-aware datetime goes in, a naive one comes back out, and every
`expires_at < datetime.now(UTC)` style comparison in auth/service.py then raises
`TypeError: can't compare offset-naive and offset-aware datetimes`. Not a quirk worth a
per-column workaround — every timestamp in this app is UTC by convention (docs/PLAN.md), so
re-attach UTC on load here, once, rather than trust the driver to preserve it.
"""
impl = DateTime(timezone=True)
cache_ok = True
def process_bind_param(self, value: datetime | None, dialect: Dialect) -> datetime | None:
if value is not None and value.tzinfo is None:
raise ValueError(
"naive datetime passed to a UTCDateTime column — always construct with "
"datetime.now(UTC), never datetime.now() or datetime.utcnow()"
)
return value
def process_result_value(self, value: Any, dialect: Dialect) -> datetime | None:
if value is not None and value.tzinfo is None:
value = value.replace(tzinfo=UTC)
return value # type: ignore[no-any-return]
class Base(DeclarativeBase):
# Every timestamp in this schema is timestamptz UTC (docs/PLAN.md's convention) — without
# this, a bare `Mapped[datetime]` infers a naive TIMESTAMP column, which then silently
# disagrees with a migration that (correctly) declares DateTime(timezone=True), and asyncpg
# rejects the mismatch at insert time. Setting it once here means every current and future
# model gets it right by default instead of each column needing to repeat it.
# Every timestamp in this schema is UTC (docs/PLAN.md's convention) — without this, a bare
# `Mapped[datetime]` infers a plain DateTime, which hits the SQLite round-trip bug above on
# every single column instead of being fixed once, here, for every current and future model.
type_annotation_map = {
datetime: DateTime(timezone=True),
datetime: UTCDateTime(),
}
+30 -22
View File
@@ -1,29 +1,33 @@
"""Identity tables: users, invites, sessions, api_tokens.
See docs/PLAN.md "Auth" and "Schema > Tables > Identity" for the design rationale, and
db.py's module docstring for why two DB roles exist. RLS policies for these tables are created
in the baseline Alembic migration (alembic/versions/0001_baseline.py), not here — SQLAlchemy
models describe columns, not database-level security policy, and keeping the policy SQL visible
and reviewable in the migration is deliberate.
See docs/PLAN.md "Auth" and "Schema > Tables > Identity" for the design rationale, and db.py's
module docstring for how isolation between users is enforced now that there's no RLS to do it at
the database level (docs/DECISIONS.md D15).
Types are deliberately dialect-generic (`sqlalchemy.Uuid`, `JSON`, plain `String` for the IP
column) rather than the `postgresql.*` variants Phase 0 originally used — this schema now targets
SQLite only, but there's no reason to hand-tie it to a Postgres-only type where a portable one
works identically. See models/base.py for why timestamps need a custom type at all.
"""
from datetime import datetime
from datetime import UTC, datetime
from uuid import UUID
from sqlalchemy import ARRAY, ForeignKey, LargeBinary, String, Text
from sqlalchemy.dialects.postgresql import INET
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy import JSON, ForeignKey, LargeBinary, String, Text, Uuid
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from velodrome.ids import new_id
from velodrome.models.base import Base
def _now_utc() -> datetime:
return datetime.now(UTC)
class User(Base):
__tablename__ = "users"
id: Mapped[UUID] = mapped_column(PGUUID(as_uuid=True), primary_key=True, default=new_id)
id: Mapped[UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=new_id)
email: Mapped[str] = mapped_column(String(320), unique=True, nullable=False)
display_name: Mapped[str] = mapped_column(String(200), nullable=False)
password_hash: Mapped[str] = mapped_column(Text, nullable=False)
@@ -32,7 +36,7 @@ class User(Base):
# Display-only, per CLAUDE.md invariant #3 — storage is always SI, this never touches a query.
unit_system: Mapped[str] = mapped_column(String(10), nullable=False, default="imperial")
is_active: Mapped[bool] = mapped_column(nullable=False, default=True)
created_at: Mapped[datetime] = mapped_column(server_default=func.now(), nullable=False)
created_at: Mapped[datetime] = mapped_column(nullable=False, default=_now_utc)
sessions: Mapped[list["Session"]] = relationship(back_populates="user")
api_tokens: Mapped[list["ApiToken"]] = relationship(back_populates="user")
@@ -41,12 +45,12 @@ class User(Base):
class Invite(Base):
__tablename__ = "invites"
id: Mapped[UUID] = mapped_column(PGUUID(as_uuid=True), primary_key=True, default=new_id)
id: Mapped[UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=new_id)
# sha256 digest of the invite code. The code itself is never stored anywhere — see
# auth/service.py. 32 bytes for sha256.
code_hash: Mapped[bytes] = mapped_column(LargeBinary(32), unique=True, nullable=False)
created_by: Mapped[UUID] = mapped_column(
PGUUID(as_uuid=True), ForeignKey("users.id"), nullable=False
Uuid(as_uuid=True), ForeignKey("users.id"), nullable=False
)
email: Mapped[str | None] = mapped_column(String(320), nullable=True)
role: Mapped[str] = mapped_column(String(20), nullable=False, default="member")
@@ -59,9 +63,9 @@ class Invite(Base):
class Session(Base):
__tablename__ = "sessions"
id: Mapped[UUID] = mapped_column(PGUUID(as_uuid=True), primary_key=True, default=new_id)
id: Mapped[UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=new_id)
user_id: Mapped[UUID] = mapped_column(
PGUUID(as_uuid=True),
Uuid(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
@@ -71,9 +75,11 @@ class Session(Base):
token_hash: Mapped[bytes] = mapped_column(LargeBinary(32), unique=True, nullable=False)
client: Mapped[str] = mapped_column(String(20), nullable=False, default="web")
user_agent: Mapped[str | None] = mapped_column(Text, nullable=True)
ip: Mapped[str | None] = mapped_column(INET, nullable=True)
created_at: Mapped[datetime] = mapped_column(server_default=func.now(), nullable=False)
last_seen_at: Mapped[datetime] = mapped_column(server_default=func.now(), nullable=False)
# Plain string, not a native INET type — SQLite has no such type, and app code never queries
# or indexes on structure within the address, only stores/displays it.
ip: Mapped[str | None] = mapped_column(String(45), nullable=True)
created_at: Mapped[datetime] = mapped_column(nullable=False, default=_now_utc)
last_seen_at: Mapped[datetime] = mapped_column(nullable=False, default=_now_utc)
expires_at: Mapped[datetime] = mapped_column(nullable=False)
revoked_at: Mapped[datetime | None] = mapped_column(nullable=True)
@@ -83,16 +89,18 @@ class Session(Base):
class ApiToken(Base):
__tablename__ = "api_tokens"
id: Mapped[UUID] = mapped_column(PGUUID(as_uuid=True), primary_key=True, default=new_id)
id: Mapped[UUID] = mapped_column(Uuid(as_uuid=True), primary_key=True, default=new_id)
user_id: Mapped[UUID] = mapped_column(
PGUUID(as_uuid=True),
Uuid(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
index=True,
)
name: Mapped[str] = mapped_column(String(200), nullable=False)
token_hash: Mapped[bytes] = mapped_column(LargeBinary(32), unique=True, nullable=False)
scopes: Mapped[list[str]] = mapped_column(ARRAY(String), nullable=False, default=list)
# JSON, not ARRAY(String) — SQLite has no array type. Stored as a JSON-encoded TEXT column;
# SQLAlchemy handles the (de)serialization transparently.
scopes: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
last_used_at: Mapped[datetime | None] = mapped_column(nullable=True)
expires_at: Mapped[datetime | None] = mapped_column(nullable=True)
revoked_at: Mapped[datetime | None] = mapped_column(nullable=True)