corvix.db

SQLAlchemy ORM models and engine helpers for PostgreSQL persistence.

Classes

Base	Shared declarative base for all ORM models.
NotificationRecordRow	Persisted notification record.
PollerStatusRow	Latest poller status (single-row table, keyed by the fixed single-user UUID).

Functions

get_database_url(→ str | None)

Return DB URL from env, supporting ${URL_ENV}_FILE Docker secret files.

Module Contents

class corvix.db.Base

Bases: sqlalchemy.orm.DeclarativeBase

Shared declarative base for all ORM models.

class corvix.db.NotificationRecordRow

Bases: Base

Persisted notification record.

__tablename__ = 'notification_records'

__table_args__

id: sqlalchemy.orm.Mapped[int]

user_id: sqlalchemy.orm.Mapped[uuid.UUID]

account_id: sqlalchemy.orm.Mapped[str]

account_label: sqlalchemy.orm.Mapped[str]

thread_id: sqlalchemy.orm.Mapped[str]

repository: sqlalchemy.orm.Mapped[str]

reason: sqlalchemy.orm.Mapped[str]

subject_title: sqlalchemy.orm.Mapped[str]

subject_type: sqlalchemy.orm.Mapped[str]

unread: sqlalchemy.orm.Mapped[bool]

updated_at: sqlalchemy.orm.Mapped[datetime.datetime]

thread_url: sqlalchemy.orm.Mapped[str | None]

web_url: sqlalchemy.orm.Mapped[str | None]

score: sqlalchemy.orm.Mapped[float]

excluded: sqlalchemy.orm.Mapped[bool]

matched_rules: sqlalchemy.orm.Mapped[list[str]]

actions_taken: sqlalchemy.orm.Mapped[list[str]]

context: sqlalchemy.orm.Mapped[dict[str, object]]

dismissed: sqlalchemy.orm.Mapped[bool]

snapshot_at: sqlalchemy.orm.Mapped[datetime.datetime]

class corvix.db.PollerStatusRow

Bases: Base

Latest poller status (single-row table, keyed by the fixed single-user UUID).

__tablename__ = 'poller_status'

user_id: sqlalchemy.orm.Mapped[uuid.UUID]

status: sqlalchemy.orm.Mapped[str]

last_poll_time: sqlalchemy.orm.Mapped[str | None]

last_error: sqlalchemy.orm.Mapped[str | None]

last_error_time: sqlalchemy.orm.Mapped[str | None]

account_errors: sqlalchemy.orm.Mapped[list[dict[str, object]] | None]

updated_at: sqlalchemy.orm.Mapped[datetime.datetime]

corvix.db.get_database_url(url_env: str = 'DATABASE_URL') → str | None

Return DB URL from env, supporting ${URL_ENV}_FILE Docker secret files.