app.db.history

Write-behind history store.

The live pipeline only enqueues rows (sync, O(1)); a background task flushes them to Postgres in batches every flush_s. The database can therefore never slow down ingest, arrival detection or forecasts.

If Postgres is unreachable the service keeps running: rows stay buffered (up to buffer_max per table, then the oldest are dropped and counted), the writer reconnects every retry_s, and /health reports degraded meanwhile (criterion 5).

Classes

History(dsn, schedules, *[, flush_s, ...])
HistoryStats(written, dropped, flush_errors)
Pool(*args, **kwargs)	The part of asyncpg.Pool the writer uses (faked in tests).

class app.db.history.History(dsn: str, schedules: dict[int, VehicleSchedule], *, flush_s: float = 1.0, batch_max: int = 5000, buffer_max: int = 200000, retry_s: float = 5.0, connect=None)

Базовые классы: object

__init__(dsn: str, schedules: dict[int, VehicleSchedule], *, flush_s: float = 1.0, batch_max: int = 5000, buffer_max: int = 200000, retry_s: float = 5.0, connect=None) → None

alert(_kind: str, a: Alert) → None

arrivals(tr_id: int, arrivals: list[Arrival]) → None

async close() → None

Last flush on shutdown, then disconnect.

async flush() → int

Write everything buffered (in batches). On failure the batch goes back to the buffer and it raises.

ping(p: Ping) → None

prediction(p: Prediction) → None

async run() → None

snapshot() → dict

class app.db.history.HistoryStats(written: 'dict[str, int]'=<factory>, dropped: 'int' = 0, flush_errors: 'int' = 0)

Базовые классы: object

__init__(written: dict[str, int]=<factory>, dropped: int = 0, flush_errors: int = 0) → None

dropped: int

flush_errors: int

written: dict[str, int]

class app.db.history.Pool(*args, **kwargs)

Базовые классы: Protocol

The part of asyncpg.Pool the writer uses (faked in tests).

__init__(*args, **kwargs)

async close() → None

async copy_records_to_table(table: str, *, records: list[tuple], columns: list[str]) → str

async execute(query: str, *args: Any) → str

async executemany(query: str, args: list[tuple]) → None