ElebobaandClaude Opus 5.5
121b607414
CI / ruff · pytest (pull_request) Successful in 45s
Two DeclarativeBases on two engines (ADR-003): identity holds direct identifiers, clinical holds token-keyed pseudonymised data plus audit_log. Portable types (JSONB variant, tz-aware UTC timestamps rejecting naive values) so tests run on SQLite. init_db() runs create_all at startup. Guardrail #5 checked by AST and import-graph tests. Refs #6 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
"""Portable column types: the same models run on Postgres (prod/compose) and SQLite (tests)."""
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import JSON, DateTime, Dialect
|
|
from sqlalchemy.dialects.postgresql import JSONB
|
|
from sqlalchemy.types import TypeDecorator
|
|
|
|
# JSONB on Postgres (indexable, compact), plain JSON elsewhere. Lists are stored as JSON instead of ARRAY
|
|
# so the schema stays portable to SQLite.
|
|
JSONType = JSON().with_variant(JSONB(), "postgresql")
|
|
|
|
|
|
class TZDateTime(TypeDecorator[datetime]):
|
|
"""Timezone-aware timestamp, stored in UTC.
|
|
|
|
Rejects naive datetimes: every business timestamp must come from `app.core.clock` (ADR-005), which
|
|
returns aware values. SQLite drops tzinfo on storage, so values read back are re-tagged as UTC.
|
|
"""
|
|
|
|
impl = DateTime(timezone=True)
|
|
cache_ok = True
|
|
|
|
def process_bind_param(self, value: datetime | None, dialect: Dialect) -> datetime | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None:
|
|
raise ValueError("naive datetime: take time from app.core.clock (ADR-005)")
|
|
return value.astimezone(UTC)
|
|
|
|
def process_result_value(self, value: Any, dialect: Dialect) -> datetime | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=UTC)
|
|
return value
|