"""SQLAlchemy ORM models and engine helpers for PostgreSQL persistence."""
from __future__ import annotations
from datetime import datetime
from uuid import UUID
from sqlalchemy import ARRAY, BigInteger, Boolean, DateTime, Float, String, Text, UniqueConstraint
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from corvix.env import get_env_value
[docs]
class Base(DeclarativeBase):
"""Shared declarative base for all ORM models."""
[docs]
class NotificationRecordRow(Base):
"""Persisted notification record."""
[docs]
__tablename__ = "notification_records"
[docs]
__table_args__ = (UniqueConstraint("user_id", "account_id", "thread_id"),)
[docs]
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
[docs]
user_id: Mapped[UUID] = mapped_column(postgresql.UUID(as_uuid=True), nullable=False)
[docs]
account_id: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
account_label: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
thread_id: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
repository: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
reason: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
subject_title: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
subject_type: Mapped[str] = mapped_column(Text, nullable=False)
[docs]
unread: Mapped[bool] = mapped_column(Boolean, nullable=False)
[docs]
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
[docs]
thread_url: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
web_url: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
score: Mapped[float] = mapped_column(Float, nullable=False)
[docs]
excluded: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
[docs]
matched_rules: Mapped[list[str]] = mapped_column(ARRAY(String), default=list)
[docs]
actions_taken: Mapped[list[str]] = mapped_column(ARRAY(String), default=list)
[docs]
context: Mapped[dict[str, object]] = mapped_column(postgresql.JSONB, nullable=False, default=dict)
[docs]
dismissed: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
[docs]
snapshot_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
[docs]
class PollerStatusRow(Base):
"""Latest poller status (single-row table, keyed by the fixed single-user UUID)."""
[docs]
__tablename__ = "poller_status"
[docs]
user_id: Mapped[UUID] = mapped_column(postgresql.UUID(as_uuid=True), primary_key=True)
[docs]
status: Mapped[str] = mapped_column(Text, nullable=False, default="unknown")
[docs]
last_poll_time: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
last_error_time: Mapped[str | None] = mapped_column(Text, nullable=True)
[docs]
account_errors: Mapped[list[dict[str, object]] | None] = mapped_column(postgresql.JSONB, nullable=True)
[docs]
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
[docs]
def get_database_url(url_env: str = "DATABASE_URL") -> str | None:
"""Return DB URL from env, supporting `${URL_ENV}_FILE` Docker secret files."""
return get_env_value(url_env)