corvix.storage

Local persistence for polled notifications.

PostgreSQL is the only supported backend; PostgresStorage implements the StorageBackend protocol used across the app.

Attributes

_NOTIFICATION_RECORD_COLUMNS
_DISMISSED_ROW_COLUMNS
_POLLER_STATUS_COLUMNS
SINGLE_USER_ID

Exceptions

StorageConfigError

Raised when required storage configuration (a database URL) is missing.

Classes

StorageBackend	Protocol for notification persistence backends.
PostgresStorage	PostgreSQL-backed notification persistence implementing StorageBackend.

Functions

create_storage(→ PostgresStorage)	Return the configured PostgreSQL storage backend.
_coerce_context(→ dict[str, object])
_coerce_string_key_dict(→ dict[str, object] | None)
_require_str(→ str)
_optional_str(→ str | None)
_require_bool(→ bool)
_require_float(→ float)
_require_datetime(→ datetime.datetime)
_parse_account_errors(...)	Parse the account_errors JSONB value into a tuple of AccountError.
_coerce_str_list(→ list[str])

Module Contents

corvix.storage._NOTIFICATION_RECORD_COLUMNS = 18

corvix.storage._DISMISSED_ROW_COLUMNS = 2

corvix.storage._POLLER_STATUS_COLUMNS = 5

corvix.storage.SINGLE_USER_ID: uuid.UUID

exception corvix.storage.StorageConfigError

Bases: RuntimeError

Raised when required storage configuration (a database URL) is missing.

class corvix.storage.StorageBackend

Bases: Protocol

Protocol for notification persistence backends.

save_records(records: list[corvix.domain.NotificationRecord], generated_at: datetime.datetime) → None

load_records() → tuple[datetime.datetime | None, list[corvix.domain.NotificationRecord]]

save_status(status: corvix.domain.PollerStatus) → None

load_status() → corvix.domain.PollerStatus

dismiss_record(thread_id: str, account_id: str = 'primary') → None

mark_record_read(thread_id: str, account_id: str = 'primary') → None

prune_orphaned_records(account_ids: collections.abc.Sequence[str]) → int

get_dismissed_notification_keys() → list[str]

get_dismissed_thread_ids() → list[str]

close() → None

__enter__() → StorageBackend

__exit__(*args: object) → None

corvix.storage.create_storage(config: corvix.config.AppConfig) → PostgresStorage

Return the configured PostgreSQL storage backend.

PostgreSQL is required in all deployments; the JSON cache is no longer used as the shared store between the poller and the web service. Raises StorageConfigError when no database URL is configured.

class corvix.storage.PostgresStorage

PostgreSQL-backed notification persistence implementing StorageBackend.

Uses psycopg (sync) so it is safe to use from CLI commands and the synchronous Litestar route handlers (sync_to_thread=False is not used with this backend — callers should run in a thread pool if needed).

A psycopg_pool.ConnectionPool is created at construction time so that TCP connections are reused across method calls rather than being opened and torn down per operation. Call close() when the storage is no longer needed, or use it as a context manager:

with PostgresStorage(connection_string=url) as storage:
    storage.save_records(...)

connection_string: str

min_pool_size: int = 1

max_pool_size: int = 10

_pool: psycopg_pool.ConnectionPool[psycopg.Connection[tuple[object, Ellipsis]]] | None = None

__post_init__() → None

close() → None

Close all pooled connections and release resources.

__enter__() → PostgresStorage

__exit__(*_args: object) → None

_connect() → contextlib.AbstractContextManager[psycopg.Connection[tuple[object, Ellipsis]]]

Return a pooled connection context-manager.

Usage is identical to the previous psycopg.connect() call:

with self._connect() as conn:
    ...

save_records(records: list[corvix.domain.NotificationRecord], generated_at: datetime.datetime) → None

Upsert records. Preserves dismissed flag on conflict.

load_records() → tuple[datetime.datetime | None, list[corvix.domain.NotificationRecord]]

Load all records ordered by snapshot_at descending.

save_status(status: corvix.domain.PollerStatus) → None

Upsert the poller status row.

load_status() → corvix.domain.PollerStatus

Load the poller status, defaulting to unknown.

dismiss_record(thread_id: str, account_id: str = 'primary') → None

Set dismissed=true for a specific account/thread id.

mark_record_read(thread_id: str, account_id: str = 'primary') → None

Set unread=false for a specific account/thread id.

prune_orphaned_records(account_ids: collections.abc.Sequence[str]) → int

Delete records whose account_id is not in account_ids.

Used to clear rows left behind when an account is removed or its id is renamed in config: such records are un-actionable in the UI (their account no longer resolves) and otherwise linger forever. account_ids must be the full set of currently-configured account IDs; accounts that merely failed to poll this cycle are still configured, so their rows are preserved. Returns the number of rows deleted.

Passing an empty sequence is a no-op (it never deletes every record), guarding against wiping the table if the configured account set is momentarily unavailable.

get_dismissed_notification_keys() → list[str]

Return account-scoped keys where dismissed=true.

get_dismissed_thread_ids() → list[str]

Return thread IDs of dismissed records.

corvix.storage._coerce_context(value: object) → dict[str, object]

corvix.storage._coerce_string_key_dict(value: object) → dict[str, object] | None

corvix.storage._require_str(value: object, field: str) → str

corvix.storage._optional_str(value: object, field: str) → str | None

corvix.storage._require_bool(value: object, field: str) → bool

corvix.storage._require_float(value: object, field: str) → float

corvix.storage._require_datetime(value: object, field: str) → datetime.datetime

corvix.storage._parse_account_errors(value: object) → tuple[corvix.domain.AccountError, Ellipsis]

Parse the account_errors JSONB value into a tuple of AccountError.

corvix.storage._coerce_str_list(value: object, field: str) → list[str]