refactor(models): declare the notification models' columns with Mapped[]

The last three model files still assigned columns as `x = Column(...)`, so
the type checker saw every field of notification providers, logs, the
digest queue, Telegram verdicts, live activities and lock-screen widgets as
a Column rather than its value. They now use Mapped[] like every other
model, `| None` exactly where the column is nullable. The schema is
unchanged (DDL and relationships compared on SQLite and Postgres). A guard
test requires Mapped[] for every model column and an annotation that
matches its nullability. Type-check baseline: 992 -> 852.

backend/app/models/notification.py

@@ -1,29 +1,37 @@
 """Notification provider and log models for push notifications."""
 
 from datetime import datetime
+from typing import TYPE_CHECKING
 
-from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String, Text
-from sqlalchemy.orm import relationship
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from backend.app.core.database import Base
 
+if TYPE_CHECKING:
+    from backend.app.models.printer import Printer
+
 
 class NotificationDigestQueue(Base):
     """Model for queuing notifications to be sent in daily digest."""
 
     __tablename__ = "notification_digest_queue"
 
-    id = Column(Integer, primary_key=True, index=True)
-    provider_id = Column(Integer, ForeignKey("notification_providers.id", ondelete="CASCADE"), nullable=False)
-    event_type = Column(String(50), nullable=False)  # print_start, print_complete, etc.
-    title = Column(String(255), nullable=False)
-    message = Column(Text, nullable=False)
-    printer_id = Column(Integer, ForeignKey("printers.id", ondelete="SET NULL"), nullable=True)
-    printer_name = Column(String(100), nullable=True)
-    created_at = Column(DateTime, default=datetime.utcnow, index=True)
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    provider_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notification_providers.id", ondelete="CASCADE"), nullable=False
+    )
+    event_type: Mapped[str] = mapped_column(String(50), nullable=False)  # print_start, print_complete, etc.
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    message: Mapped[str] = mapped_column(Text, nullable=False)
+    printer_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("printers.id", ondelete="SET NULL"), nullable=True
+    )
+    printer_name: Mapped[str | None] = mapped_column(String(100), nullable=True)
+    created_at: Mapped[datetime | None] = mapped_column(DateTime, default=datetime.utcnow, index=True)
 
     # Relationships
-    provider = relationship("NotificationProvider", back_populates="digest_queue")
+    provider: Mapped["NotificationProvider"] = relationship(back_populates="digest_queue")
 
 
 class NotificationLog(Base):
@@ -31,19 +39,25 @@ class NotificationLog(Base):
 
     __tablename__ = "notification_logs"
 
-    id = Column(Integer, primary_key=True, index=True)
-    provider_id = Column(Integer, ForeignKey("notification_providers.id", ondelete="CASCADE"), nullable=False)
-    event_type = Column(String(50), nullable=False)  # print_start, print_complete, etc.
-    title = Column(String(255), nullable=False)
-    message = Column(Text, nullable=False)
-    success = Column(Boolean, default=True)
-    error_message = Column(Text, nullable=True)
-    printer_id = Column(Integer, ForeignKey("printers.id", ondelete="SET NULL"), nullable=True)
-    printer_name = Column(String(100), nullable=True)  # Store name in case printer is deleted
-    created_at = Column(DateTime, default=datetime.utcnow, index=True)
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    provider_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notification_providers.id", ondelete="CASCADE"), nullable=False
+    )
+    event_type: Mapped[str] = mapped_column(String(50), nullable=False)  # print_start, print_complete, etc.
+    title: Mapped[str] = mapped_column(String(255), nullable=False)
+    message: Mapped[str] = mapped_column(Text, nullable=False)
+    success: Mapped[bool | None] = mapped_column(Boolean, default=True)
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    printer_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("printers.id", ondelete="SET NULL"), nullable=True
+    )
+    printer_name: Mapped[str | None] = mapped_column(
+        String(100), nullable=True
+    )  # Store name in case printer is deleted
+    created_at: Mapped[datetime | None] = mapped_column(DateTime, default=datetime.utcnow, index=True)
 
     # Relationships
-    provider = relationship("NotificationProvider", back_populates="logs")
+    provider: Mapped["NotificationProvider"] = relationship(back_populates="logs")
 
 
 class NotificationProvider(Base):
@@ -51,120 +65,154 @@ class NotificationProvider(Base):
 
     __tablename__ = "notification_providers"
 
-    id = Column(Integer, primary_key=True, index=True)
-    name = Column(String(100), nullable=False)  # User-defined name
-    provider_type = Column(String(50), nullable=False)  # callmebot, ntfy, pushover, telegram, email
-    enabled = Column(Boolean, default=True)
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    name: Mapped[str] = mapped_column(String(100), nullable=False)  # User-defined name
+    provider_type: Mapped[str] = mapped_column(String(50), nullable=False)  # callmebot, ntfy, pushover, telegram, email
+    enabled: Mapped[bool | None] = mapped_column(Boolean, default=True)
 
     # Provider-specific configuration stored as JSON string
-    config = Column(Text, nullable=False)
+    config: Mapped[str] = mapped_column(Text, nullable=False)
 
     # Attach a camera snapshot when one was captured for the event. Defaults
     # True so ntfy/Pushover/Telegram/Discord keep behaving as they always
     # did; this is the opt-out for whoever doesn't want it.
-    attach_photo = Column(Boolean, default=True)
+    attach_photo: Mapped[bool | None] = mapped_column(Boolean, default=True)
 
     # Event triggers - print lifecycle
-    on_print_start = Column(Boolean, default=False)
-    on_print_complete = Column(Boolean, default=True)
-    on_print_failed = Column(Boolean, default=True)
-    on_print_stopped = Column(Boolean, default=True)  # User cancelled/stopped print
-    on_print_progress = Column(Boolean, default=False)  # 25%, 50%, 75% milestones
-    on_print_missing_spool_assignment = Column(Boolean, default=False)  # Print started with unassigned required tray(s)
-    on_billing_charge_failed = Column(Boolean, default=True)  # A completed/stopped print could not be charged
+    on_print_start: Mapped[bool | None] = mapped_column(Boolean, default=False)
+    on_print_complete: Mapped[bool | None] = mapped_column(Boolean, default=True)
+    on_print_failed: Mapped[bool | None] = mapped_column(Boolean, default=True)
+    on_print_stopped: Mapped[bool | None] = mapped_column(Boolean, default=True)  # User cancelled/stopped print
+    on_print_progress: Mapped[bool | None] = mapped_column(Boolean, default=False)  # 25%, 50%, 75% milestones
+    on_print_missing_spool_assignment: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # Print started with unassigned required tray(s)
+    on_billing_charge_failed: Mapped[bool | None] = mapped_column(
+        Boolean, default=True
+    )  # A completed/stopped print could not be charged
 
     # Event triggers - printer status
-    on_printer_offline = Column(Boolean, default=False)
-    on_printer_error = Column(Boolean, default=False)  # AMS issues, etc.
-    on_ai_failure_detection = Column(Boolean, default=False)  # Obico spaghetti / failure detection (#1794)
-    on_filament_low = Column(Boolean, default=False)
-    on_maintenance_due = Column(Boolean, default=False)  # Maintenance reminder
+    on_printer_offline: Mapped[bool | None] = mapped_column(Boolean, default=False)
+    on_printer_error: Mapped[bool | None] = mapped_column(Boolean, default=False)  # AMS issues, etc.
+    on_ai_failure_detection: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # Obico spaghetti / failure detection (#1794)
+    on_filament_low: Mapped[bool | None] = mapped_column(Boolean, default=False)
+    on_maintenance_due: Mapped[bool | None] = mapped_column(Boolean, default=False)  # Maintenance reminder
 
     # Event triggers - AMS environmental alarms (regular AMS with 4 slots)
-    on_ams_humidity_high = Column(Boolean, default=False)  # AMS humidity above threshold
-    on_ams_temperature_high = Column(Boolean, default=False)  # AMS temperature above threshold
+    on_ams_humidity_high: Mapped[bool | None] = mapped_column(Boolean, default=False)  # AMS humidity above threshold
+    on_ams_temperature_high: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # AMS temperature above threshold
     # Auto-drying gave up on a unit (#2770). Defaults True: it reports that
     # Bambuddy has stopped acting, which nothing else in the UI would say.
-    on_ams_drying_suspended = Column(Boolean, default=True)
+    on_ams_drying_suspended: Mapped[bool | None] = mapped_column(Boolean, default=True)
 
     # Event triggers - AMS-HT environmental alarms (single slot heated AMS)
-    on_ams_ht_humidity_high = Column(Boolean, default=False)  # AMS-HT humidity above threshold
-    on_ams_ht_temperature_high = Column(Boolean, default=False)  # AMS-HT temperature above threshold
+    on_ams_ht_humidity_high: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # AMS-HT humidity above threshold
+    on_ams_ht_temperature_high: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # AMS-HT temperature above threshold
 
     # Event triggers - Home Assistant sensors bound to a printer (#1148)
-    on_ha_sensor_alert = Column(Boolean, default=False)  # Bound HA sensor entered its alert state
+    on_ha_sensor_alert: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # Bound HA sensor entered its alert state
 
     # Event triggers - Home Assistant sensors bound to a storage location (#2824)
     # Its own column rather than reusing on_ha_sensor_alert above: that one can
     # be scoped to a single printer, and a location alert has no printer to
     # scope by, so sharing it would leak drybox alerts to a provider narrowed
     # to one printer's sensors.
-    on_location_ha_sensor_alert = Column(Boolean, default=False)
+    on_location_ha_sensor_alert: Mapped[bool | None] = mapped_column(Boolean, default=False)
 
     # Event triggers - Build plate detection
-    on_plate_not_empty = Column(Boolean, default=True)  # Objects detected on plate before print
+    on_plate_not_empty: Mapped[bool | None] = mapped_column(
+        Boolean, default=True
+    )  # Objects detected on plate before print
     # Off by default: fires after every print, alongside the print-complete alert (#2525)
-    on_plate_clear_required = Column(Boolean, default=False)  # Print ended, queue gated until plate is confirmed clear
+    on_plate_clear_required: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # Print ended, queue gated until plate is confirmed clear
 
     # Print asked for an outcome verdict (#1898). Defaults ON: it only ever
     # fires for prints where the user opted in per-job, so the provider-level
     # toggle exists to silence a channel, not to enable the feature.
-    on_print_confirm_request = Column(Boolean, default=True)
+    on_print_confirm_request: Mapped[bool | None] = mapped_column(Boolean, default=True)
     # How a Telegram provider collects that verdict (#3046): inline URL
     # buttons ("buttons", the original behaviour), a thumbs-up/down reaction
     # on the message ("reactions"), or both. Reactions need no inbound
     # connectivity — the poller fetches them — so they work without an
     # external_url. Ignored for every other provider type.
-    telegram_verdict_mode = Column(String(16), default="buttons")
+    telegram_verdict_mode: Mapped[str | None] = mapped_column(String(16), default="buttons")
 
     # Event triggers - Bed cooled after print
-    on_bed_cooled = Column(Boolean, default=False)  # Bed cooled below threshold after print
-    on_first_layer_complete = Column(Boolean, default=False)  # First layer finished printing
+    on_bed_cooled: Mapped[bool | None] = mapped_column(Boolean, default=False)  # Bed cooled below threshold after print
+    on_first_layer_complete: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # First layer finished printing
     # Messages another application sends through Bambuddy (POST /notifications/app-message),
     # e.g. Bambuddy Orders' "an order needs you". Off by default: nothing new arrives on upgrade.
-    on_app_message = Column(Boolean, default=False)
+    on_app_message: Mapped[bool | None] = mapped_column(Boolean, default=False)
 
     # Event triggers - Inventory stock alerts
-    on_stock_reorder_alert = Column(Boolean, default=False)  # SKU hits reorder point
-    on_stock_break_alert = Column(Boolean, default=False)  # Stock will run out before replenishment
+    on_stock_reorder_alert: Mapped[bool | None] = mapped_column(Boolean, default=False)  # SKU hits reorder point
+    on_stock_break_alert: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # Stock will run out before replenishment
 
     # Event triggers - Print queue
-    on_queue_job_added = Column(Boolean, default=False)  # Job added to queue
-    on_queue_job_assigned = Column(Boolean, default=False)  # Model-based job assigned to printer
-    on_queue_job_started = Column(Boolean, default=False)  # Queue job started printing
-    on_queue_job_waiting = Column(Boolean, default=True)  # Job waiting for filament or printer
-    on_queue_job_skipped = Column(Boolean, default=True)  # Job skipped (previous print failed)
-    on_queue_job_failed = Column(Boolean, default=True)  # Job failed to start
-    on_queue_completed = Column(Boolean, default=False)  # All pending jobs finished
+    on_queue_job_added: Mapped[bool | None] = mapped_column(Boolean, default=False)  # Job added to queue
+    on_queue_job_assigned: Mapped[bool | None] = mapped_column(
+        Boolean, default=False
+    )  # Model-based job assigned to printer
+    on_queue_job_started: Mapped[bool | None] = mapped_column(Boolean, default=False)  # Queue job started printing
+    on_queue_job_waiting: Mapped[bool | None] = mapped_column(
+        Boolean, default=True
+    )  # Job waiting for filament or printer
+    on_queue_job_skipped: Mapped[bool | None] = mapped_column(
+        Boolean, default=True
+    )  # Job skipped (previous print failed)
+    on_queue_job_failed: Mapped[bool | None] = mapped_column(Boolean, default=True)  # Job failed to start
+    on_queue_completed: Mapped[bool | None] = mapped_column(Boolean, default=False)  # All pending jobs finished
 
     # Quiet hours (do not disturb)
-    quiet_hours_enabled = Column(Boolean, default=False)
-    quiet_hours_start = Column(String(5), nullable=True)  # HH:MM format, e.g., "22:00"
-    quiet_hours_end = Column(String(5), nullable=True)  # HH:MM format, e.g., "07:00"
+    quiet_hours_enabled: Mapped[bool | None] = mapped_column(Boolean, default=False)
+    quiet_hours_start: Mapped[str | None] = mapped_column(String(5), nullable=True)  # HH:MM format, e.g., "22:00"
+    quiet_hours_end: Mapped[str | None] = mapped_column(String(5), nullable=True)  # HH:MM format, e.g., "07:00"
 
     # Daily digest (batch notifications into a single daily summary)
-    daily_digest_enabled = Column(Boolean, default=False)
-    daily_digest_time = Column(String(5), nullable=True)  # HH:MM format, e.g., "08:00"
+    daily_digest_enabled: Mapped[bool | None] = mapped_column(Boolean, default=False)
+    daily_digest_time: Mapped[str | None] = mapped_column(String(5), nullable=True)  # HH:MM format, e.g., "08:00"
 
     # Optional: Link to specific printer (NULL = all printers)
-    printer_id = Column(Integer, ForeignKey("printers.id", ondelete="SET NULL"), nullable=True)
+    printer_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("printers.id", ondelete="SET NULL"), nullable=True
+    )
 
     # Status tracking
-    last_success = Column(DateTime, nullable=True)
-    last_error = Column(Text, nullable=True)
-    last_error_at = Column(DateTime, nullable=True)
+    last_success: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    last_error_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
 
     # Timestamps
-    created_at = Column(DateTime, default=datetime.utcnow)
-    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
+    created_at: Mapped[datetime | None] = mapped_column(DateTime, default=datetime.utcnow)
+    updated_at: Mapped[datetime | None] = mapped_column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
 
     # Relationships
-    printer = relationship("Printer", back_populates="notification_providers")
-    logs = relationship("NotificationLog", back_populates="provider", cascade="all, delete-orphan")
-    digest_queue = relationship("NotificationDigestQueue", back_populates="provider", cascade="all, delete-orphan")
+    printer: Mapped["Printer | None"] = relationship(back_populates="notification_providers")
+    logs: Mapped[list["NotificationLog"]] = relationship(back_populates="provider", cascade="all, delete-orphan")
+    digest_queue: Mapped[list["NotificationDigestQueue"]] = relationship(
+        back_populates="provider", cascade="all, delete-orphan"
+    )
     # ORM-level cascade like the two above: SQLite ships with foreign keys
     # off, so the column's ON DELETE CASCADE alone would leave rows behind.
-    pending_verdicts = relationship("TelegramPendingVerdict", back_populates="provider", cascade="all, delete-orphan")
+    pending_verdicts: Mapped[list["TelegramPendingVerdict"]] = relationship(
+        back_populates="provider", cascade="all, delete-orphan"
+    )
 
 
 class TelegramPendingVerdict(Base):
@@ -178,22 +226,26 @@ class TelegramPendingVerdict(Base):
 
     __tablename__ = "telegram_pending_verdicts"
 
-    id = Column(Integer, primary_key=True, index=True)
-    provider_id = Column(Integer, ForeignKey("notification_providers.id", ondelete="CASCADE"), nullable=False)
-    chat_id = Column(String(64), nullable=False)
-    message_id = Column(Integer, nullable=False)
-    archive_id = Column(Integer, ForeignKey("print_archives.id", ondelete="CASCADE"), nullable=False)
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    provider_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notification_providers.id", ondelete="CASCADE"), nullable=False
+    )
+    chat_id: Mapped[str] = mapped_column(String(64), nullable=False)
+    message_id: Mapped[int] = mapped_column(Integer, nullable=False)
+    archive_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("print_archives.id", ondelete="CASCADE"), nullable=False
+    )
     # The archive's confirm_token when the prompt went out. A reaction counts
     # only while that is still the archive's live token, i.e. exactly as long
     # as the prompt's one-tap links would: a reprint clears the token, so an
     # earlier run's message can no longer answer the new run.
-    confirm_token = Column(String(64), nullable=True)
+    confirm_token: Mapped[str | None] = mapped_column(String(64), nullable=True)
     # sendPhoto messages carry the prompt as a caption, which is edited with a
     # different Bot API method than a plain text message.
-    has_caption = Column(Boolean, default=False)
+    has_caption: Mapped[bool | None] = mapped_column(Boolean, default=False)
     # The Markdown we sent, so the confirmation edit can append to it — the
     # reaction update does not carry the message body.
-    message_text = Column(Text, nullable=True)
-    created_at = Column(DateTime, default=datetime.utcnow, index=True)
+    message_text: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_at: Mapped[datetime | None] = mapped_column(DateTime, default=datetime.utcnow, index=True)
 
-    provider = relationship("NotificationProvider", back_populates="pending_verdicts")
+    provider: Mapped["NotificationProvider"] = relationship(back_populates="pending_verdicts")

backend/app/models/notification_live_activity.py

@@ -2,7 +2,8 @@
 
 from datetime import datetime
 
-from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
+from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
 
 from backend.app.core.database import Base
 
@@ -11,24 +12,24 @@ class NotificationLiveActivity(Base):
     __tablename__ = "notification_live_activities"
     __table_args__ = (UniqueConstraint("provider_id", "printer_id", "print_key"),)
 
-    id = Column(Integer, primary_key=True)
-    provider_id = Column(
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    provider_id: Mapped[int] = mapped_column(
         Integer, ForeignKey("notification_providers.id", ondelete="CASCADE"), nullable=False, index=True
     )
     # Keep ownership after printer deletion until the worker ends the remote tile.
-    printer_id = Column(Integer, nullable=False, index=True)
-    print_key = Column(String(200), nullable=False)
-    job_key = Column(String(160), nullable=True)
-    print_name = Column(String(255), nullable=False, default="")
-    eta_seconds = Column(Integer, nullable=True)
-    eta_deadline = Column(DateTime, nullable=True)
-    credential_key = Column(String(64), nullable=False)
-    activity_id = Column(String(16), nullable=True)
-    state = Column(String(20), nullable=False, default="pending")
-    end_reason = Column(String(40), nullable=True)
-    failures = Column(Integer, nullable=False, default=0)
-    content = Column(Text, nullable=False, default="{}")
-    expires_at = Column(DateTime, nullable=True)
-    next_attempt_at = Column(DateTime, nullable=True)
-    last_sent_at = Column(DateTime, nullable=True)
-    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
+    printer_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
+    print_key: Mapped[str] = mapped_column(String(200), nullable=False)
+    job_key: Mapped[str | None] = mapped_column(String(160), nullable=True)
+    print_name: Mapped[str] = mapped_column(String(255), nullable=False, default="")
+    eta_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    eta_deadline: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    credential_key: Mapped[str] = mapped_column(String(64), nullable=False)
+    activity_id: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    state: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
+    end_reason: Mapped[str | None] = mapped_column(String(40), nullable=True)
+    failures: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    content: Mapped[str] = mapped_column(Text, nullable=False, default="{}")
+    expires_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    last_sent_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.utcnow)

backend/app/models/notification_lock_screen_widget.py

@@ -2,7 +2,8 @@
 
 from datetime import datetime
 
-from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
+from sqlalchemy import DateTime, ForeignKey, Integer, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
 
 from backend.app.core.database import Base
 
@@ -11,17 +12,17 @@ class NotificationLockScreenWidget(Base):
     __tablename__ = "notification_lock_screen_widgets"
     __table_args__ = (UniqueConstraint("provider_id", "printer_id"),)
 
-    id = Column(Integer, primary_key=True)
-    provider_id = Column(
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    provider_id: Mapped[int] = mapped_column(
         Integer, ForeignKey("notification_providers.id", ondelete="CASCADE"), nullable=False, index=True
     )
     # A removed printer must leave ownership intact until the remote DELETE succeeds.
-    printer_id = Column(Integer, nullable=False, index=True)
-    credential_key = Column(String(64), nullable=False)
-    widget_id = Column(String(16), nullable=True)
-    state = Column(String(20), nullable=False, default="pending")
-    content = Column(Text, nullable=False, default="{}")
-    failures = Column(Integer, nullable=False, default=0)
-    next_attempt_at = Column(DateTime, nullable=True)
-    last_sent_at = Column(DateTime, nullable=True)
-    created_at = Column(DateTime, nullable=False, default=datetime.utcnow)
+    printer_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
+    credential_key: Mapped[str] = mapped_column(String(64), nullable=False)
+    widget_id: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    state: Mapped[str] = mapped_column(String(20), nullable=False, default="pending")
+    content: Mapped[str] = mapped_column(Text, nullable=False, default="{}")
+    failures: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    last_sent_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=datetime.utcnow)

backend/tests/unit/test_code_quality.py

@@ -289,6 +289,23 @@ class TestLogErrorPatterns:
         assert not capture_logs.has_errors(), f"Errors during MQTT processing:\n{capture_logs.format_errors()}"
 
 
+def _load_all_models():
+    """Import every model module and return the declarative Base.
+
+    The package __init__ does not import every model module; without this a
+    table outside it would never be checked.
+    """
+    import importlib
+    import pkgutil
+
+    import backend.app.models
+    from backend.app.core.database import Base
+
+    for module in pkgutil.iter_modules(backend.app.models.__path__):
+        importlib.import_module(f"backend.app.models.{module.name}")
+    return Base
+
+
 class TestDateTimeColumns:
     """No model column may be DateTime(timezone=True).
 
@@ -299,19 +316,9 @@ class TestDateTimeColumns:
     """
 
     def test_no_timezone_aware_datetime_columns(self):
-        import importlib
-        import pkgutil
-
         from sqlalchemy import DateTime
 
-        import backend.app.models
-        from backend.app.core.database import Base
-
-        # The package __init__ does not import every model module, so load them
-        # all here; otherwise a table outside it would never be checked.
-        for module in pkgutil.iter_modules(backend.app.models.__path__):
-            importlib.import_module(f"backend.app.models.{module.name}")
-
+        Base = _load_all_models()
         offenders = [
             f"{table.name}.{column.name}"
             for table in Base.metadata.sorted_tables
@@ -319,3 +326,43 @@ class TestDateTimeColumns:
             if isinstance(column.type, DateTime) and column.type.timezone
         ]
         assert not offenders, f"DateTime(timezone=True) columns: {offenders}"
+
+
+class TestMappedColumns:
+    """Every model column is declared with Mapped[], and its type says whether it can be NULL.
+
+    A column assigned as ``x = Column(...)`` reaches the type checker as a
+    Column, not as the value it holds, so every use of it is unchecked (or a
+    false error in the type-check baseline). And a Mapped[] annotation that
+    disagrees with the column's nullability lies to every reader: Mapped[str]
+    on a nullable column hides the None the code must handle; Mapped[str | None]
+    on a NOT NULL column asks for checks that can never matter.
+    """
+
+    def test_columns_are_mapped_and_nullability_matches(self):
+        import typing
+
+        Base = _load_all_models()
+        models = {mapper.class_.__name__: mapper.class_ for mapper in Base.registry.mappers}
+        legacy, mismatched = [], []
+        for mapper in Base.registry.mappers:
+            hints: dict = {}
+            for klass in reversed(mapper.class_.__mro__):
+                hints.update(typing.get_type_hints(klass, localns=models))
+            for column in mapper.local_table.columns:
+                if column.primary_key:
+                    continue
+                key = mapper.get_property_by_column(column).key
+                name = f"{mapper.class_.__name__}.{key}"
+                hint = hints.get(key)
+                if hint is None:
+                    legacy.append(name)
+                    continue
+                optional = type(None) in typing.get_args(typing.get_args(hint)[0])
+                if optional != column.nullable:
+                    mismatched.append(
+                        f"{name}: annotated {'optional' if optional else 'required'}, "
+                        f"column {'nullable' if column.nullable else 'NOT NULL'}"
+                    )
+        assert not legacy, f"declare these with Mapped[...] = mapped_column(...): {legacy}"
+        assert not mismatched, f"Mapped[] annotation disagrees with the column: {mismatched}"