From 121b607414035b042275d2e029d81ae1fb296fda Mon Sep 17 00:00:00 2001 From: Lacky1234union Date: Sat, 3 Oct 2026 14:58:25 +0300 Subject: [PATCH] =?UTF-8?q?feat(backend):=20data=20model=20=E2=80=94=20ide?= =?UTF-8?q?ntity=20/=20clinical=20/=20audit,=20repositories,=20create=5Fal?= =?UTF-8?q?l?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/db/audit.py | 41 ++++++ app/db/bootstrap.py | 17 +++ app/db/clinical.py | 194 ++++++++++++++++++++++++++ app/db/identity.py | 63 +++++++++ app/db/identity_repo.py | 136 ++++++++++++++++++ app/db/repo.py | 298 ++++++++++++++++++++++++++++++++++++++++ app/db/session.py | 89 ++++++++++++ app/db/types.py | 37 +++++ app/main.py | 13 +- tests/test_db.py | 287 ++++++++++++++++++++++++++++++++++++++ 10 files changed, 1174 insertions(+), 1 deletion(-) create mode 100644 app/db/audit.py create mode 100644 app/db/bootstrap.py create mode 100644 app/db/clinical.py create mode 100644 app/db/identity.py create mode 100644 app/db/identity_repo.py create mode 100644 app/db/repo.py create mode 100644 app/db/session.py create mode 100644 app/db/types.py create mode 100644 tests/test_db.py diff --git a/app/db/audit.py b/app/db/audit.py new file mode 100644 index 0000000..e31cceb --- /dev/null +++ b/app/db/audit.py @@ -0,0 +1,41 @@ +"""Audit log (guardrail #6): every doctor access to patient data and every safety-critical system action. + +Lives in the clinical DB (ADR-003) and references patients only by token. +""" + +from datetime import datetime +from typing import Any + +from sqlalchemy import String, Text +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import Mapped, mapped_column + +from app.db.clinical import TOKEN_LEN, BigId, ClinicalBase +from app.db.types import JSONType, TZDateTime + + +class AuditLog(ClinicalBase): + __tablename__ = "audit_log" + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + ts: Mapped[datetime] = mapped_column(TZDateTime, index=True) + actor: Mapped[str] = mapped_column(String(64)) # doctor_id | system + action: Mapped[str] = mapped_column(Text) # view_patient | ack_alert | crisis_acute | export ... + token: Mapped[str | None] = mapped_column(String(TOKEN_LEN), index=True) + details: Mapped[dict[str, Any] | None] = mapped_column(JSONType) + + +async def audit( + session: AsyncSession, + *, + actor: str, + action: str, + token: str | None, + details: dict[str, Any] | None = None, + ts: datetime, +) -> AuditLog: + """Append an audit record in the caller's transaction. `details` must not contain PII or raw text.""" + entry = AuditLog(ts=ts, actor=actor, action=action, token=token, details=details) + session.add(entry) + await session.flush() + return entry diff --git a/app/db/bootstrap.py b/app/db/bootstrap.py new file mode 100644 index 0000000..d157c3c --- /dev/null +++ b/app/db/bootstrap.py @@ -0,0 +1,17 @@ +"""Schema bootstrap: `create_all` on both databases at startup (ADR-003: no Alembic in the MVP). + +The only module (besides registration code) that touches both schemas — and only their metadata. +""" + +from app.db import audit as _audit # noqa: F401 (registers audit_log in clinical metadata) +from app.db.clinical import ClinicalBase +from app.db.identity import IdentityBase +from app.db.session import clinical_engine, identity_engine + + +async def init_db() -> None: + """Create missing tables in the identity and clinical databases. Idempotent.""" + async with identity_engine().begin() as conn: + await conn.run_sync(IdentityBase.metadata.create_all) + async with clinical_engine().begin() as conn: + await conn.run_sync(ClinicalBase.metadata.create_all) diff --git a/app/db/clinical.py b/app/db/clinical.py new file mode 100644 index 0000000..9b9478d --- /dev/null +++ b/app/db/clinical.py @@ -0,0 +1,194 @@ +"""Clinical DB schema: pseudonymised health data. SPEC §5, ADR-002, ADR-003, ADR-006. + +Every row is keyed by `token` (32 hex chars of HMAC(PSEUDO_KEY, SNILS)), never by name/SNILS/phone/chat id. +Free text is stored only after `scrub()` (`*_scrubbed` columns). This module must not import +`app.db.identity` (guardrail #5, checked in tests/test_db.py); `audit_log` also lives in this DB. + +Business timestamps have no server defaults: callers pass time from `app.core.clock` (ADR-005), so the +demo virtual clock is reflected consistently in data, rules and the dashboard. +""" + +from datetime import date, datetime +from typing import Any + +from sqlalchemy import ( + BigInteger, + Boolean, + Date, + Float, + ForeignKey, + Index, + Integer, + SmallInteger, + String, + Text, + UniqueConstraint, +) +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + +from app.db.types import JSONType, TZDateTime + +TOKEN_LEN = 32 +# BIGSERIAL on Postgres; SQLite autoincrements only INTEGER PRIMARY KEY. +BigId = BigInteger().with_variant(Integer(), "sqlite") + + +class ClinicalBase(DeclarativeBase): + """Metadata of the clinical DB (incl. audit_log); never mixed with identity metadata.""" + + +class PatientProfile(ClinicalBase): + """Clinical view of a patient: no name, SNILS or contacts — only what monitoring needs.""" + + __tablename__ = "patient_profiles" + + token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True) + doctor_id: Mapped[str] = mapped_column(String(64), index=True) + modules: Mapped[list[str]] = mapped_column(JSONType, default=list) # veteran | depression | family + therapy_start: Mapped[date | None] = mapped_column(Date) + next_visit: Mapped[date | None] = mapped_column(Date) + birth_year: Mapped[int | None] = mapped_column(SmallInteger) + sex: Mapped[str | None] = mapped_column(String(1)) + # Set by rules (e.g. after an acute episode): denser check-ins and lower silence thresholds. + enhanced_monitoring: Mapped[bool] = mapped_column(Boolean, default=False) + # Cache for the dashboard list; the source of truth is open alerts (colour is derived, ADR-006). + status_color: Mapped[str | None] = mapped_column(String(8)) + + +class ItemResponse(ClinicalBase): + """Single scale item answer (PHQ-9, C-SSRS...) from buttons, LLM extraction or clinician.""" + + __tablename__ = "item_responses" + __table_args__ = (Index("ix_item_responses_token_ts", "token", "ts"),) + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(TOKEN_LEN)) + ts: Mapped[datetime] = mapped_column(TZDateTime) + scale: Mapped[str] = mapped_column(String(16)) # phq9 | gad7 | pcl5 | cssrs | auditc ... + item_id: Mapped[str] = mapped_column(String(32)) + score: Mapped[int | None] = mapped_column(SmallInteger) + confidence: Mapped[float | None] = mapped_column(Float) + evidence: Mapped[str | None] = mapped_column(Text) # scrubbed phrase supporting an LLM-extracted score + source: Mapped[str] = mapped_column(String(16)) # llm | button | clinician + + +class ScaleResult(ClinicalBase): + __tablename__ = "scale_results" + __table_args__ = (Index("ix_scale_results_token_ts", "token", "ts"),) + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(TOKEN_LEN)) + ts: Mapped[datetime] = mapped_column(TZDateTime) + scale: Mapped[str] = mapped_column(String(16)) + period_start: Mapped[date | None] = mapped_column(Date) + period_end: Mapped[date | None] = mapped_column(Date) + total: Mapped[int] = mapped_column(SmallInteger) + severity: Mapped[str] = mapped_column(String(32)) + complete: Mapped[bool] = mapped_column(Boolean) + # PHQ-9 item 9 (suicidal ideation) kept separately: any value >= 1 escalates regardless of the total. + item9: Mapped[int | None] = mapped_column(SmallInteger) + # False = unscheduled administration triggered by a rule (e.g. mood <= 3 for three days, ADR-002). + scheduled: Mapped[bool] = mapped_column(Boolean, default=True) + flags: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict) + + +class DailyCheckin(ClinicalBase): + """One row per patient per (virtual) day, filled in several steps of the daily dialog (ADR-002).""" + + __tablename__ = "daily_checkins" + __table_args__ = (UniqueConstraint("token", "day", name="uq_daily_checkins_token_day"),) + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(TOKEN_LEN)) + day: Mapped[date] = mapped_column(Date) + mood: Mapped[int | None] = mapped_column(SmallInteger) # 0..10 + med_taken: Mapped[bool | None] = mapped_column(Boolean) + sleep_bucket: Mapped[str | None] = mapped_column(String(16)) + alcohol_bucket: Mapped[str | None] = mapped_column(String(16)) + side_effects: Mapped[list[str]] = mapped_column(JSONType, default=list) + quit_intent: Mapped[bool | None] = mapped_column(Boolean) # wants to stop medication — doctor flag + nightmares: Mapped[bool | None] = mapped_column(Boolean) + free_text_scrubbed: Mapped[str | None] = mapped_column(Text) + updated_at: Mapped[datetime] = mapped_column(TZDateTime) + + +class Prescription(ClinicalBase): + __tablename__ = "prescriptions" + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(TOKEN_LEN), index=True) + drug: Mapped[str] = mapped_column(String(128)) # INN (МНН) + dose: Mapped[str] = mapped_column(String(64)) + schedule: Mapped[dict[str, Any]] = mapped_column(JSONType) # {"times": ["08:00", "20:00"]} + start_date: Mapped[date | None] = mapped_column(Date) + end_date: Mapped[date | None] = mapped_column(Date) + + +class Dose(ClinicalBase): + __tablename__ = "doses" + __table_args__ = (Index("ix_doses_token_planned", "token", "planned_at"),) + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(TOKEN_LEN)) + prescription_id: Mapped[int] = mapped_column(ForeignKey("prescriptions.id")) + planned_at: Mapped[datetime] = mapped_column(TZDateTime) + status: Mapped[str] = mapped_column(String(16), default="pending") # taken | missed | skipped | pending + answered_at: Mapped[datetime | None] = mapped_column(TZDateTime) + side_effects: Mapped[str | None] = mapped_column(Text) + + +class Alert(ClinicalBase): + """Output of the deterministic rules engine (ADR-007). Level is stored, UI colour is derived (ADR-006).""" + + __tablename__ = "alerts" + __table_args__ = (Index("ix_alerts_token_ts", "token", "ts"),) + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(TOKEN_LEN)) + ts: Mapped[datetime] = mapped_column(TZDateTime) + level: Mapped[str] = mapped_column(String(16)) # acute | high | moderate | info (AlertLevel name) + code: Mapped[str] = mapped_column(String(64)) # rule id from config/rules.yaml + reason: Mapped[str] = mapped_column(Text) + params: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict) + acked_by: Mapped[str | None] = mapped_column(String(64)) + acked_at: Mapped[datetime | None] = mapped_column(TZDateTime) + + +class Message(ClinicalBase): + __tablename__ = "messages" + __table_args__ = (Index("ix_messages_token_ts", "token", "ts"),) + + id: Mapped[int] = mapped_column(BigId, primary_key=True, autoincrement=True) + token: Mapped[str] = mapped_column(String(TOKEN_LEN)) + ts: Mapped[datetime] = mapped_column(TZDateTime) + direction: Mapped[str] = mapped_column(String(3)) # in | out + text_scrubbed: Mapped[str] = mapped_column(Text) + fsm_state: Mapped[str | None] = mapped_column(String(64)) + + +class DialogState(ClinicalBase): + """FSM state of the dialog engine, persisted instead of Redis (ADR-003).""" + + __tablename__ = "dialog_state" + + token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True) + state: Mapped[str] = mapped_column(String(64)) + data: Mapped[dict[str, Any]] = mapped_column(JSONType, default=dict) + updated_at: Mapped[datetime] = mapped_column(TZDateTime) + + +class SafetyPlan(ClinicalBase): + __tablename__ = "safety_plans" + + token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True) + steps: Mapped[list[Any]] = mapped_column(JSONType, default=list) + updated_at: Mapped[datetime] = mapped_column(TZDateTime) + + +class ClockState(ClinicalBase): + """Single row (id=1): virtual clock offset so the demo timeline survives restarts (ADR-005).""" + + __tablename__ = "clock_state" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + offset_seconds: Mapped[float] = mapped_column(Float, default=0.0) diff --git a/app/db/identity.py b/app/db/identity.py new file mode 100644 index 0000000..c54d521 --- /dev/null +++ b/app/db/identity.py @@ -0,0 +1,63 @@ +"""Identity DB schema: direct identifiers (152-FZ personal data). SPEC §5, ADR-003. + +Why a separate database: a leak or over-broad query on the clinical side must not reveal who the patient +is. Names, SNILS, phones and Telegram chat ids live only here; the clinical DB stores health data keyed by +`token` = HMAC(PSEUDO_KEY, SNILS) (`app.core.privacy.pseudo_id`). Without the key and this DB the clinical +data is pseudonymised. In compose each DB has its own Postgres role with no access to the other one. + +Only registration and doctor-notification code may import this module (guardrail #5). +""" + +from datetime import datetime + +from sqlalchemy import BigInteger, Boolean, ForeignKey, SmallInteger, String, Text +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + +from app.db.types import TZDateTime + +TOKEN_LEN = 32 + + +class IdentityBase(DeclarativeBase): + """Metadata of the identity DB only; never mixed with clinical metadata.""" + + +class Patient(IdentityBase): + __tablename__ = "patients" + + token: Mapped[str] = mapped_column(String(TOKEN_LEN), primary_key=True) + display_name: Mapped[str] = mapped_column(Text) + snils: Mapped[str] = mapped_column(String(14), unique=True) + phone: Mapped[str | None] = mapped_column(String(32)) + tg_chat_id: Mapped[int | None] = mapped_column(BigInteger, unique=True) + birth_year: Mapped[int | None] = mapped_column(SmallInteger) + sex: Mapped[str | None] = mapped_column(String(1)) # "m" | "f" + clinic_id: Mapped[str] = mapped_column(String(64)) + doctor_id: Mapped[str] = mapped_column(String(64), index=True) + # NULL consent_at = no consent: processing of health data is forbidden (152-FZ art. 10). + consent_at: Mapped[datetime | None] = mapped_column(TZDateTime) + consent_ver: Mapped[str | None] = mapped_column(String(32)) + revoked_at: Mapped[datetime | None] = mapped_column(TZDateTime) + created_at: Mapped[datetime] = mapped_column(TZDateTime) + + +class Doctor(IdentityBase): + __tablename__ = "doctors" + + doctor_id: Mapped[str] = mapped_column(String(64), primary_key=True) + full_name: Mapped[str] = mapped_column(Text) + clinic_id: Mapped[str] = mapped_column(String(64)) + notify_chat: Mapped[int | None] = mapped_column(BigInteger) + on_call: Mapped[bool] = mapped_column(Boolean, default=False) + + +class ConnectCode(IdentityBase): + """One-time code the doctor gives the patient (QR) to bind the bot chat to the patient record.""" + + __tablename__ = "connect_codes" + + code: Mapped[str] = mapped_column(String(32), primary_key=True) + token: Mapped[str] = mapped_column(ForeignKey("patients.token"), index=True) + doctor_id: Mapped[str] = mapped_column(String(64)) + created_at: Mapped[datetime] = mapped_column(TZDateTime) + used_at: Mapped[datetime | None] = mapped_column(TZDateTime) diff --git a/app/db/identity_repo.py b/app/db/identity_repo.py new file mode 100644 index 0000000..234574c --- /dev/null +++ b/app/db/identity_repo.py @@ -0,0 +1,136 @@ +"""Identity-DB helpers: registration, consent, connect codes, doctor lookup. + +Only registration / notification code may use this module. Functions flush but never commit and take time +as an explicit argument (ADR-005). +""" + +from collections.abc import Sequence +from datetime import datetime + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.identity import ConnectCode, Doctor, Patient + + +async def create_patient( + session: AsyncSession, + *, + token: str, + display_name: str, + snils: str, + clinic_id: str, + doctor_id: str, + now: datetime, + phone: str | None = None, + birth_year: int | None = None, + sex: str | None = None, +) -> Patient: + patient = Patient( + token=token, + display_name=display_name, + snils=snils, + clinic_id=clinic_id, + doctor_id=doctor_id, + created_at=now, + phone=phone, + birth_year=birth_year, + sex=sex, + ) + session.add(patient) + await session.flush() + return patient + + +async def get_patient(session: AsyncSession, token: str) -> Patient | None: + return await session.get(Patient, token) + + +async def get_patient_by_snils(session: AsyncSession, snils: str) -> Patient | None: + return await session.scalar(select(Patient).where(Patient.snils == snils)) + + +async def get_patient_by_chat(session: AsyncSession, tg_chat_id: int) -> Patient | None: + return await session.scalar(select(Patient).where(Patient.tg_chat_id == tg_chat_id)) + + +async def bind_chat(session: AsyncSession, *, token: str, tg_chat_id: int) -> Patient | None: + patient = await session.get(Patient, token) + if patient is not None: + patient.tg_chat_id = tg_chat_id + await session.flush() + return patient + + +async def record_consent(session: AsyncSession, *, token: str, version: str, now: datetime) -> Patient | None: + patient = await session.get(Patient, token) + if patient is not None: + patient.consent_at = now + patient.consent_ver = version + patient.revoked_at = None + await session.flush() + return patient + + +async def revoke_consent(session: AsyncSession, *, token: str, now: datetime) -> Patient | None: + patient = await session.get(Patient, token) + if patient is not None: + patient.revoked_at = now + await session.flush() + return patient + + +def has_active_consent(patient: Patient) -> bool: + """Health data may be processed only with consent given and not revoked (152-FZ art. 10).""" + return patient.consent_at is not None and patient.revoked_at is None + + +async def upsert_doctor( + session: AsyncSession, + *, + doctor_id: str, + full_name: str, + clinic_id: str, + notify_chat: int | None = None, + on_call: bool = False, +) -> Doctor: + doctor = await session.get(Doctor, doctor_id) + if doctor is None: + doctor = Doctor(doctor_id=doctor_id) + session.add(doctor) + doctor.full_name = full_name + doctor.clinic_id = clinic_id + doctor.notify_chat = notify_chat + doctor.on_call = on_call + await session.flush() + return doctor + + +async def get_doctor(session: AsyncSession, doctor_id: str) -> Doctor | None: + return await session.get(Doctor, doctor_id) + + +async def doctors_on_call(session: AsyncSession, clinic_id: str | None = None) -> Sequence[Doctor]: + stmt = select(Doctor).where(Doctor.on_call.is_(True)) + if clinic_id is not None: + stmt = stmt.where(Doctor.clinic_id == clinic_id) + return (await session.scalars(stmt.order_by(Doctor.doctor_id))).all() + + +async def create_connect_code( + session: AsyncSession, *, code: str, token: str, doctor_id: str, now: datetime +) -> ConnectCode: + row = ConnectCode(code=code, token=token, doctor_id=doctor_id, created_at=now) + session.add(row) + await session.flush() + return row + + +async def redeem_connect_code(session: AsyncSession, *, code: str, now: datetime) -> str | None: + """Mark a one-time connect code as used and return the patient token; None if unknown or already used.""" + row = await session.get(ConnectCode, code) + if row is None or row.used_at is not None: + return None + row.used_at = now + await session.flush() + return row.token diff --git a/app/db/repo.py b/app/db/repo.py new file mode 100644 index 0000000..8bbbfb9 --- /dev/null +++ b/app/db/repo.py @@ -0,0 +1,298 @@ +"""Thin clinical-DB repository helpers. Must not import `app.db.identity` (guardrail #5). + +Functions work inside the caller's session (`get_clinical_session()`), flush but never commit, and take +time as an explicit argument (ADR-005). +""" + +from collections.abc import Sequence +from datetime import date, datetime +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.types import AlertLevel, ScrubbedText +from app.db.audit import audit +from app.db.clinical import ( + Alert, + ClockState, + DailyCheckin, + DialogState, + ItemResponse, + Message, + PatientProfile, + ScaleResult, +) + +_CHECKIN_FIELDS = frozenset( + { + "mood", + "med_taken", + "sleep_bucket", + "alcohol_bucket", + "side_effects", + "quit_intent", + "nightmares", + "free_text_scrubbed", + } +) +_PROFILE_FIELDS = frozenset( + { + "doctor_id", + "modules", + "therapy_start", + "next_visit", + "birth_year", + "sex", + "enhanced_monitoring", + "status_color", + } +) + + +# --- patient profiles ------------------------------------------------------------------------------- + + +async def get_profile(session: AsyncSession, token: str) -> PatientProfile | None: + return await session.get(PatientProfile, token) + + +async def upsert_profile(session: AsyncSession, *, token: str, **fields: Any) -> PatientProfile: + """Create or update a profile; only the passed fields are changed.""" + unknown = set(fields) - _PROFILE_FIELDS + if unknown: + raise ValueError(f"unknown profile fields: {sorted(unknown)}") + profile = await session.get(PatientProfile, token) + if profile is None: + profile = PatientProfile(token=token, modules=[], enhanced_monitoring=False) + session.add(profile) + for key, value in fields.items(): + setattr(profile, key, value) + await session.flush() + return profile + + +async def patients_of_doctor(session: AsyncSession, doctor_id: str) -> Sequence[PatientProfile]: + rows = await session.scalars( + select(PatientProfile).where(PatientProfile.doctor_id == doctor_id).order_by(PatientProfile.token) + ) + return rows.all() + + +# --- daily check-ins and scales --------------------------------------------------------------------- + + +async def upsert_checkin( + session: AsyncSession, *, token: str, day: date, now: datetime, **fields: Any +) -> DailyCheckin: + """Merge answers into the (token, day) check-in. + + The daily dialog fills the row step by step (mood, then medication, then the module question), so + only fields passed and not None overwrite stored values. + """ + unknown = set(fields) - _CHECKIN_FIELDS + if unknown: + raise ValueError(f"unknown check-in fields: {sorted(unknown)}") + row = await session.scalar( + select(DailyCheckin).where(DailyCheckin.token == token, DailyCheckin.day == day) + ) + if row is None: + row = DailyCheckin(token=token, day=day, side_effects=[], updated_at=now) + session.add(row) + for key, value in fields.items(): + if value is not None: + setattr(row, key, value) + row.updated_at = now + await session.flush() + return row + + +async def list_checkins( + session: AsyncSession, token: str, since: date | None = None +) -> Sequence[DailyCheckin]: + stmt = select(DailyCheckin).where(DailyCheckin.token == token) + if since is not None: + stmt = stmt.where(DailyCheckin.day >= since) + return (await session.scalars(stmt.order_by(DailyCheckin.day))).all() + + +async def add_item_response( + session: AsyncSession, + *, + token: str, + scale: str, + item_id: str, + score: int | None, + source: str, + ts: datetime, + confidence: float | None = None, + evidence: ScrubbedText | None = None, +) -> ItemResponse: + row = ItemResponse( + token=token, + scale=scale, + item_id=item_id, + score=score, + source=source, + ts=ts, + confidence=confidence, + evidence=evidence, + ) + session.add(row) + await session.flush() + return row + + +async def add_scale_result( + session: AsyncSession, + *, + token: str, + scale: str, + total: int, + severity: str, + complete: bool, + ts: datetime, + item9: int | None = None, + scheduled: bool = True, + flags: dict[str, Any] | None = None, + period_start: date | None = None, + period_end: date | None = None, +) -> ScaleResult: + row = ScaleResult( + token=token, + scale=scale, + total=total, + severity=severity, + complete=complete, + ts=ts, + item9=item9, + scheduled=scheduled, + flags=flags or {}, + period_start=period_start, + period_end=period_end, + ) + session.add(row) + await session.flush() + return row + + +async def list_scale_results( + session: AsyncSession, token: str, scale: str | None = None +) -> Sequence[ScaleResult]: + stmt = select(ScaleResult).where(ScaleResult.token == token) + if scale is not None: + stmt = stmt.where(ScaleResult.scale == scale) + return (await session.scalars(stmt.order_by(ScaleResult.ts))).all() + + +# --- alerts ----------------------------------------------------------------------------------------- + + +async def add_alert( + session: AsyncSession, + *, + token: str, + level: AlertLevel, + code: str, + reason: str, + ts: datetime, + params: dict[str, Any] | None = None, +) -> Alert: + row = Alert(token=token, level=level.name.lower(), code=code, reason=reason, ts=ts, params=params or {}) + session.add(row) + await session.flush() + return row + + +async def open_alert_codes(session: AsyncSession, token: str) -> set[str]: + """Codes of unacknowledged alerts — lets rules avoid raising the same alert twice.""" + rows = await session.scalars(select(Alert.code).where(Alert.token == token, Alert.acked_at.is_(None))) + return set(rows.all()) + + +async def list_alerts( + session: AsyncSession, *, token: str | None = None, open_only: bool = False +) -> Sequence[Alert]: + stmt = select(Alert) + if token is not None: + stmt = stmt.where(Alert.token == token) + if open_only: + stmt = stmt.where(Alert.acked_at.is_(None)) + return (await session.scalars(stmt.order_by(Alert.ts.desc(), Alert.id.desc()))).all() + + +async def ack_alert(session: AsyncSession, *, alert_id: int, doctor_id: str, ts: datetime) -> Alert | None: + """Doctor acknowledges an alert; the access is written to audit_log in the same transaction. + + Re-acking is a no-op (first acknowledgement wins, no duplicate audit record). Unknown id -> None. + """ + alert = await session.get(Alert, alert_id) + if alert is None or alert.acked_at is not None: + return alert + alert.acked_by = doctor_id + alert.acked_at = ts + await audit( + session, + actor=doctor_id, + action="ack_alert", + token=alert.token, + details={"alert_id": alert.id, "code": alert.code, "level": alert.level}, + ts=ts, + ) + return alert + + +# --- dialog ----------------------------------------------------------------------------------------- + + +async def get_dialog_state(session: AsyncSession, token: str) -> DialogState | None: + return await session.get(DialogState, token) + + +async def set_dialog_state( + session: AsyncSession, *, token: str, state: str, data: dict[str, Any], now: datetime +) -> DialogState: + row = await session.get(DialogState, token) + if row is None: + row = DialogState(token=token, state=state, data=data, updated_at=now) + session.add(row) + else: + row.state = state + row.data = dict(data) # new object so SQLAlchemy sees the JSON change + row.updated_at = now + await session.flush() + return row + + +async def add_message( + session: AsyncSession, + *, + token: str, + direction: str, + text: ScrubbedText, + ts: datetime, + fsm_state: str | None = None, +) -> Message: + if direction not in ("in", "out"): + raise ValueError("direction must be 'in' or 'out'") + row = Message(token=token, direction=direction, text_scrubbed=text, ts=ts, fsm_state=fsm_state) + session.add(row) + await session.flush() + return row + + +# --- demo clock persistence (ADR-005) ---------------------------------------------------------------- + + +async def load_clock_offset(session: AsyncSession) -> float: + row = await session.get(ClockState, 1) + return row.offset_seconds if row else 0.0 + + +async def save_clock_offset(session: AsyncSession, offset_seconds: float) -> None: + row = await session.get(ClockState, 1) + if row is None: + session.add(ClockState(id=1, offset_seconds=offset_seconds)) + else: + row.offset_seconds = offset_seconds + await session.flush() diff --git a/app/db/session.py b/app/db/session.py new file mode 100644 index 0000000..c66d439 --- /dev/null +++ b/app/db/session.py @@ -0,0 +1,89 @@ +"""Engines and sessions for the two databases (ADR-003). + +This module knows nothing about models: clinical code can import it without pulling in the identity +schema (guardrail #5). Schema creation lives in `app.db.bootstrap`. +""" + +from collections.abc import AsyncIterator +from contextlib import AbstractAsyncContextManager, asynccontextmanager + +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.pool import StaticPool + +from app.config import get_settings + +_engines: dict[str, AsyncEngine] = {} +_makers: dict[str, async_sessionmaker[AsyncSession]] = {} +_urls: dict[str, str] = {} + + +def _make_engine(url: str) -> AsyncEngine: + kwargs: dict[str, object] = {} + if url.startswith("postgresql"): + # Fail fast when the DB is unreachable instead of hanging on asyncpg's 60 s default. + kwargs = {"pool_pre_ping": True, "connect_args": {"timeout": 5}} + elif url.startswith("sqlite") and (":memory:" in url or url.rstrip("/").endswith(":")): + # In-memory SQLite (tests): one shared connection, otherwise every connection sees an empty DB. + kwargs = {"poolclass": StaticPool} + return create_async_engine(url, **kwargs) + + +def _url(name: str) -> str: + if name in _urls: + return _urls[name] + settings = get_settings() + return settings.identity_db_url if name == "identity" else settings.clinical_db_url + + +def _engine(name: str) -> AsyncEngine: + if name not in _engines: + _engines[name] = _make_engine(_url(name)) + _makers[name] = async_sessionmaker(_engines[name], expire_on_commit=False) + return _engines[name] + + +def identity_engine() -> AsyncEngine: + return _engine("identity") + + +def clinical_engine() -> AsyncEngine: + return _engine("clinical") + + +async def configure(*, identity_url: str | None = None, clinical_url: str | None = None) -> None: + """Override DB URLs (tests, scripts). Disposes engines created so far; new ones are created lazily.""" + await dispose() + _urls.clear() + if identity_url: + _urls["identity"] = identity_url + if clinical_url: + _urls["clinical"] = clinical_url + + +@asynccontextmanager +async def _session(name: str) -> AsyncIterator[AsyncSession]: + _engine(name) + async with _makers[name]() as session: + try: + yield session + await session.commit() + except BaseException: + await session.rollback() + raise + + +def get_identity_session() -> AbstractAsyncContextManager[AsyncSession]: + """Unit of work on the identity DB: commits on success, rolls back on error.""" + return _session("identity") + + +def get_clinical_session() -> AbstractAsyncContextManager[AsyncSession]: + """Unit of work on the clinical DB (incl. audit_log): commits on success, rolls back on error.""" + return _session("clinical") + + +async def dispose() -> None: + for engine in _engines.values(): + await engine.dispose() + _engines.clear() + _makers.clear() diff --git a/app/db/types.py b/app/db/types.py new file mode 100644 index 0000000..e17dbd2 --- /dev/null +++ b/app/db/types.py @@ -0,0 +1,37 @@ +"""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 diff --git a/app/main.py b/app/main.py index 316a337..a118dc5 100644 --- a/app/main.py +++ b/app/main.py @@ -5,8 +5,11 @@ from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from fastapi import FastAPI +from sqlalchemy.exc import SQLAlchemyError from app.config import get_settings +from app.db.bootstrap import init_db +from app.db.session import dispose # Logs carry only tokens, states and event codes — never PII or raw message text. logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") @@ -17,8 +20,16 @@ log = logging.getLogger("app") async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]: settings = get_settings() log.info("startup llm_provider=%s demo_clock=%s", settings.llm_provider, settings.demo_clock) - # DB init (#6), scheduler (#13) and Telegram bot (#11) hook in here. + try: + await init_db() + log.info("db schema ready") + except (OSError, SQLAlchemyError) as exc: + # In compose Postgres is required and healthy before the app starts; this branch is for running the + # API (and its tests) without a database. Endpoints touching the DB will fail until it is reachable. + log.error("db init failed, continuing without db: %s", type(exc).__name__) + # Scheduler (#13) and Telegram bot (#11) hook in here. yield + await dispose() app = FastAPI(title="PTSRaiAgent API", version="0.1.0", lifespan=lifespan) diff --git a/tests/test_db.py b/tests/test_db.py new file mode 100644 index 0000000..9dd80aa --- /dev/null +++ b/tests/test_db.py @@ -0,0 +1,287 @@ +"""Data model tests on two separate in-memory SQLite databases (identity / clinical). + +Set TEST_PG_IDENTITY_URL / TEST_PG_CLINICAL_URL to run the same suite against compose Postgres. +""" + +import ast +import os +import subprocess +import sys +from collections.abc import AsyncIterator, Callable +from datetime import UTC, date, datetime, timedelta +from pathlib import Path + +import pytest +from sqlalchemy import inspect, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncEngine + +from app.core.types import AlertLevel, ScrubbedText +from app.db import identity_repo, repo, session +from app.db.audit import AuditLog, audit +from app.db.bootstrap import init_db +from app.db.clinical import ClinicalBase, DailyCheckin, Dose, Prescription +from app.db.identity import IdentityBase + +DB_DIR = Path(__file__).resolve().parent.parent / "app" / "db" +T0 = datetime(2026, 10, 3, 9, 0, tzinfo=UTC) +TOKEN = "a" * 32 +TOKEN2 = "b" * 32 + + +@pytest.fixture +async def dbs() -> AsyncIterator[None]: + await session.configure( + identity_url=os.environ.get("TEST_PG_IDENTITY_URL", "sqlite+aiosqlite:///:memory:"), + clinical_url=os.environ.get("TEST_PG_CLINICAL_URL", "sqlite+aiosqlite:///:memory:"), + ) + if "TEST_PG_IDENTITY_URL" in os.environ: + async with session.identity_engine().begin() as conn: + await conn.run_sync(IdentityBase.metadata.drop_all) + async with session.clinical_engine().begin() as conn: + await conn.run_sync(ClinicalBase.metadata.drop_all) + await init_db() + yield + await session.configure() + + +async def _tables(engine_fn: Callable[[], AsyncEngine]) -> set[str]: + async with engine_fn().connect() as conn: + return set(await conn.run_sync(lambda c: inspect(c).get_table_names())) + + +async def test_create_all_puts_tables_in_the_right_db(dbs: None) -> None: + identity_tables = await _tables(session.identity_engine) + clinical_tables = await _tables(session.clinical_engine) + assert identity_tables == {"patients", "doctors", "connect_codes"} + assert { + "patient_profiles", + "item_responses", + "scale_results", + "daily_checkins", + "prescriptions", + "doses", + "alerts", + "messages", + "dialog_state", + "safety_plans", + "clock_state", + "audit_log", + } == clinical_tables + await init_db() # idempotent + + +async def test_identity_roundtrip_and_connect_code(dbs: None) -> None: + async with session.get_identity_session() as s: + await identity_repo.upsert_doctor( + s, doctor_id="d1", full_name="Доктор Тест", clinic_id="c1", on_call=True + ) + await identity_repo.create_patient( + s, + token=TOKEN, + display_name="Пациент", + snils="000-000-000 00", + clinic_id="c1", + doctor_id="d1", + now=T0, + ) + await identity_repo.create_connect_code(s, code="ABC123", token=TOKEN, doctor_id="d1", now=T0) + async with session.get_identity_session() as s: + assert await identity_repo.redeem_connect_code(s, code="ABC123", now=T0) == TOKEN + assert await identity_repo.redeem_connect_code(s, code="ABC123", now=T0) is None + assert await identity_repo.redeem_connect_code(s, code="NOPE", now=T0) is None + await identity_repo.bind_chat(s, token=TOKEN, tg_chat_id=42) + patient = await identity_repo.record_consent(s, token=TOKEN, version="v1", now=T0) + assert patient is not None and identity_repo.has_active_consent(patient) + async with session.get_identity_session() as s: + patient = await identity_repo.get_patient_by_chat(s, 42) + assert patient is not None and patient.token == TOKEN + assert patient.created_at == T0 and patient.created_at.tzinfo is not None + await identity_repo.revoke_consent(s, token=TOKEN, now=T0 + timedelta(days=1)) + assert not identity_repo.has_active_consent(patient) + assert [d.doctor_id for d in await identity_repo.doctors_on_call(s)] == ["d1"] + + +async def test_snils_and_chat_unique(dbs: None) -> None: + async with session.get_identity_session() as s: + await identity_repo.create_patient( + s, token=TOKEN, display_name="A", snils="111", clinic_id="c", doctor_id="d", now=T0 + ) + with pytest.raises(IntegrityError): + async with session.get_identity_session() as s: + await identity_repo.create_patient( + s, token=TOKEN2, display_name="B", snils="111", clinic_id="c", doctor_id="d", now=T0 + ) + async with session.get_identity_session() as s: + await identity_repo.create_patient( + s, token=TOKEN2, display_name="B", snils="222", clinic_id="c", doctor_id="d", now=T0 + ) + await identity_repo.bind_chat(s, token=TOKEN, tg_chat_id=7) + with pytest.raises(IntegrityError): + async with session.get_identity_session() as s: + await identity_repo.bind_chat(s, token=TOKEN2, tg_chat_id=7) + + +async def test_checkin_upsert_merges_and_is_unique_per_day(dbs: None) -> None: + day = date(2026, 10, 3) + async with session.get_clinical_session() as s: + await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, mood=4) + await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, med_taken=True, side_effects=["тошнота"]) + await repo.upsert_checkin(s, token=TOKEN, day=day + timedelta(days=1), now=T0, mood=3) + async with session.get_clinical_session() as s: + rows = await repo.list_checkins(s, TOKEN) + assert len(rows) == 2 + assert (rows[0].mood, rows[0].med_taken, rows[0].side_effects) == (4, True, ["тошнота"]) + assert [r.day for r in await repo.list_checkins(s, TOKEN, since=day + timedelta(days=1))] == [ + day + timedelta(days=1) + ] + with pytest.raises(ValueError): + await repo.upsert_checkin(s, token=TOKEN, day=day, now=T0, bogus=1) + with pytest.raises(IntegrityError): + async with session.get_clinical_session() as s: + s.add(DailyCheckin(token=TOKEN, day=day, side_effects=[], updated_at=T0)) + + +async def test_scales_profiles_prescriptions(dbs: None) -> None: + async with session.get_clinical_session() as s: + await repo.upsert_profile(s, token=TOKEN, doctor_id="d1", modules=["veteran", "depression"]) + await repo.upsert_profile(s, token=TOKEN2, doctor_id="d2") + await repo.upsert_profile(s, token=TOKEN, next_visit=date(2026, 10, 20)) + await repo.add_item_response( + s, token=TOKEN, scale="phq9", item_id="9", score=1, source="button", ts=T0 + ) + await repo.add_scale_result( + s, + token=TOKEN, + scale="phq9", + total=16, + severity="moderately_severe", + complete=True, + ts=T0, + item9=1, + scheduled=False, + flags={"trigger": "low_mood_3d"}, + ) + rx = Prescription(token=TOKEN, drug="sertraline", dose="50 mg", schedule={"times": ["08:00"]}) + s.add(rx) + await s.flush() + s.add(Dose(token=TOKEN, prescription_id=rx.id, planned_at=T0)) + async with session.get_clinical_session() as s: + profiles = await repo.patients_of_doctor(s, "d1") + assert [(p.token, p.modules, p.next_visit) for p in profiles] == [ + (TOKEN, ["veteran", "depression"], date(2026, 10, 20)) + ] + (result,) = await repo.list_scale_results(s, TOKEN, "phq9") + assert (result.item9, result.scheduled, result.flags) == (1, False, {"trigger": "low_mood_3d"}) + dose = await s.scalar(select(Dose)) + assert dose is not None and dose.status == "pending" and dose.planned_at == T0 + + +async def test_ack_alert_writes_audit(dbs: None) -> None: + async with session.get_clinical_session() as s: + alert = await repo.add_alert( + s, + token=TOKEN, + level=AlertLevel.ACUTE, + code="phq9_item9", + reason="п.9 ≥ 1", + ts=T0, + params={"item9": 1}, + ) + await repo.add_alert(s, token=TOKEN, level=AlertLevel.INFO, code="silence", reason="молчание", ts=T0) + alert_id = alert.id + assert await repo.open_alert_codes(s, TOKEN) == {"phq9_item9", "silence"} + later = T0 + timedelta(minutes=5) + async with session.get_clinical_session() as s: + acked = await repo.ack_alert(s, alert_id=alert_id, doctor_id="d1", ts=later) + assert acked is not None and acked.acked_by == "d1" + await repo.ack_alert(s, alert_id=alert_id, doctor_id="d2", ts=later) # no-op + assert await repo.ack_alert(s, alert_id=9999, doctor_id="d1", ts=later) is None + async with session.get_clinical_session() as s: + assert await repo.open_alert_codes(s, TOKEN) == {"silence"} + (stored,) = [a for a in await repo.list_alerts(s, token=TOKEN) if a.id == alert_id] + assert (stored.level, stored.acked_by, stored.acked_at) == ("acute", "d1", later) + logs = (await s.scalars(select(AuditLog))).all() + assert [(a.actor, a.action, a.token, a.ts) for a in logs] == [("d1", "ack_alert", TOKEN, later)] + assert logs[0].details == {"alert_id": alert_id, "code": "phq9_item9", "level": "acute"} + + +async def test_audit_helper(dbs: None) -> None: + async with session.get_clinical_session() as s: + await audit(s, actor="system", action="crisis_acute", token=TOKEN, details=None, ts=T0) + async with session.get_clinical_session() as s: + entry = await s.scalar(select(AuditLog)) + assert entry is not None and entry.action == "crisis_acute" + + +async def test_dialog_state_messages_clock(dbs: None) -> None: + async with session.get_clinical_session() as s: + assert await repo.get_dialog_state(s, TOKEN) is None + await repo.set_dialog_state(s, token=TOKEN, state="checkin_mood", data={"step": 1}, now=T0) + await repo.add_message( + s, token=TOKEN, direction="in", text=ScrubbedText("ок"), ts=T0, fsm_state="idle" + ) + with pytest.raises(ValueError): + await repo.add_message(s, token=TOKEN, direction="x", text=ScrubbedText("ок"), ts=T0) + assert await repo.load_clock_offset(s) == 0.0 + await repo.save_clock_offset(s, 86400.0) + async with session.get_clinical_session() as s: + await repo.set_dialog_state(s, token=TOKEN, state="checkin_med", data={"step": 2}, now=T0) + await repo.save_clock_offset(s, 2 * 86400.0) + async with session.get_clinical_session() as s: + state = await repo.get_dialog_state(s, TOKEN) + assert state is not None and (state.state, state.data) == ("checkin_med", {"step": 2}) + assert await repo.load_clock_offset(s) == 2 * 86400.0 + + +async def test_naive_datetime_rejected(dbs: None) -> None: + with pytest.raises(Exception, match="naive datetime"): + async with session.get_clinical_session() as s: + await repo.add_alert( + s, token=TOKEN, level=AlertLevel.INFO, code="x", reason="x", ts=datetime(2026, 1, 1) + ) + + +async def test_session_rolls_back_on_error(dbs: None) -> None: + with pytest.raises(RuntimeError): + async with session.get_clinical_session() as s: + await repo.upsert_profile(s, token=TOKEN, doctor_id="d1") + raise RuntimeError + async with session.get_clinical_session() as s: + assert await repo.get_profile(s, TOKEN) is None + + +# --- guardrail #5: clinical code never touches identity ---------------------------------------------- + +CLINICAL_MODULES = ["clinical.py", "repo.py", "audit.py", "session.py", "types.py"] + + +def _imported_modules(path: Path) -> set[str]: + tree = ast.parse(path.read_text(encoding="utf-8")) + names: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + names.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + names.add(node.module or "") + names.update(f"{node.module}.{alias.name}" for alias in node.names) + return names + + +@pytest.mark.parametrize("filename", CLINICAL_MODULES) +def test_clinical_code_does_not_import_identity(filename: str) -> None: + imported = _imported_modules(DB_DIR / filename) + assert not [m for m in imported if "identity" in m], f"{filename} imports identity: {imported}" + + +def test_clinical_import_graph_excludes_identity() -> None: + code = "import sys, app.db.repo, app.db.audit; print('app.db.identity' in sys.modules)" + out = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True, check=True, cwd=DB_DIR.parents[1] + ) + assert out.stdout.strip() == "False" + + +def test_metadata_is_separate() -> None: + assert not set(IdentityBase.metadata.tables) & set(ClinicalBase.metadata.tables) + assert "audit_log" in ClinicalBase.metadata.tables