Files
PTSRaiAgent/app/db/audit.py
ElebobaandClaude Opus 5.5 121b607414
CI / ruff · pytest (pull_request) Successful in 45s
feat(backend): data model — identity / clinical / audit, repositories, create_all
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>
2026-10-03 14:58:26 +03:00

42 lines
1.4 KiB
Python

"""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