ElebobaandClaude Opus 5.5
b4731fb1d6
CI / ruff · pytest (pull_request) Successful in 22s
Freeze interfaces between lanes A (bot), B (backend), C (dashboard) so three developers work in parallel without touching each other's files: dialog Event/Reply + engine registry with echo stub, integration protocols with in-memory defaults, MonitoringService with noop stub, /api/v1 schemas with fixture-backed routes, frozen main.py wiring, config/modules.yaml + due_scales(), DAR-5, tzdata, Andrey trajectory, lanes in WORKFLOW, ADR-010. Refs #34 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
115 lines
3.5 KiB
Python
115 lines
3.5 KiB
Python
"""Integration protocols (frozen, #34). Demo uses mocks; prod swaps implementations (SPEC §9).
|
|
|
|
Lane B implements the real mocks in `app/integrations/*` (#14) and registers them via `set_*`.
|
|
The in-memory defaults below keep every lane unblocked and testable before #14 lands.
|
|
"""
|
|
|
|
from collections import defaultdict
|
|
from datetime import date
|
|
from typing import Literal, Protocol
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.dialog.contracts import Reply
|
|
|
|
AlertLevelName = Literal["info", "moderate", "high", "acute"]
|
|
|
|
|
|
class Prescription(BaseModel):
|
|
drug: str # INN, e.g. "сертралин"
|
|
dose: str # "50 мг"
|
|
schedule: dict[str, list[str]] = Field(default_factory=lambda: {"times": ["08:00"]})
|
|
instructions: str = "" # "утром после еды"
|
|
start_date: date | None = None
|
|
end_date: date | None = None
|
|
|
|
|
|
class VerifiedPerson(BaseModel):
|
|
token: str
|
|
doctor_id: str
|
|
|
|
|
|
class DrugSection(BaseModel):
|
|
inn: str
|
|
section: str # side_effects | onset | alcohol | discontinuation | ...
|
|
text: str
|
|
source: str # reference to the ГРЛС instruction
|
|
|
|
|
|
class PrescriptionSource(Protocol): # МИС / ЕГИСЗ
|
|
async def get_prescriptions(self, token: str) -> list[Prescription]: ...
|
|
|
|
|
|
class IdentityProvider(Protocol): # ЕСИА; demo: connect code from the doctor
|
|
async def verify(self, auth_code: str) -> VerifiedPerson | None: ...
|
|
|
|
|
|
class Notifier(Protocol): # doctor notifications: dashboard feed + Telegram chat / MAX / МИС
|
|
async def send(self, doctor_id: str, level: AlertLevelName, text: str, token: str | None = None) -> None: ...
|
|
|
|
|
|
class DrugInfo(Protocol): # ГРЛС
|
|
async def sections(self, inn: str) -> list[DrugSection]: ...
|
|
|
|
|
|
class Outbox(Protocol):
|
|
"""Delivers engine replies initiated by the system (reminders, check-ins) to the patient's channel."""
|
|
|
|
async def push(self, token: str, replies: list[Reply]) -> None: ...
|
|
|
|
|
|
# ----------------------------------------------------------------- in-memory defaults
|
|
|
|
|
|
class MemoryOutbox:
|
|
"""Default outbox: per-token queue polled by the web demo console (`drain`)."""
|
|
|
|
def __init__(self) -> None:
|
|
self._q: dict[str, list[Reply]] = defaultdict(list)
|
|
|
|
async def push(self, token: str, replies: list[Reply]) -> None:
|
|
self._q[token].extend(replies)
|
|
|
|
def drain(self, token: str) -> list[Reply]:
|
|
out, self._q[token] = self._q[token], []
|
|
return out
|
|
|
|
|
|
class MemoryNotifier:
|
|
"""Default notifier: keeps notifications in memory (no PII in `text` — reason codes only)."""
|
|
|
|
def __init__(self) -> None:
|
|
self.sent: list[tuple[str, str, str, str | None]] = []
|
|
|
|
async def send(self, doctor_id: str, level: AlertLevelName, text: str, token: str | None = None) -> None:
|
|
self.sent.append((doctor_id, level, text, token))
|
|
|
|
|
|
class EmptyPrescriptions:
|
|
async def get_prescriptions(self, token: str) -> list[Prescription]:
|
|
return []
|
|
|
|
|
|
class NoDrugInfo:
|
|
async def sections(self, inn: str) -> list[DrugSection]:
|
|
return []
|
|
|
|
|
|
class RejectIdentity:
|
|
async def verify(self, auth_code: str) -> VerifiedPerson | None:
|
|
return None
|
|
|
|
|
|
class Integrations:
|
|
"""Registry. Lane B replaces fields at startup (`app.integrations.setup()`, #14)."""
|
|
|
|
def __init__(self) -> None:
|
|
self.outbox: Outbox = MemoryOutbox()
|
|
self.notifier: Notifier = MemoryNotifier()
|
|
self.prescriptions: PrescriptionSource = EmptyPrescriptions()
|
|
self.drug_info: DrugInfo = NoDrugInfo()
|
|
self.identity: IdentityProvider = RejectIdentity()
|
|
|
|
|
|
integrations = Integrations()
|