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>
42 lines
1.4 KiB
Python
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
|