"""Import log, audit log, and notification models.""" from __future__ import annotations import datetime as dt from typing import Optional from sqlalchemy import ( Boolean, CheckConstraint, DateTime, ForeignKey, Index, Integer, String, Text, func, ) from sqlalchemy.dialects.mysql import JSON from sqlalchemy.orm import Mapped, mapped_column, relationship from app.database import Base class ImportLog(Base): __tablename__ = "import_log" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) filename: Mapped[str] = mapped_column(String(255), nullable=False) import_type: Mapped[str] = mapped_column(String(50), nullable=False) cases_imported: Mapped[int] = mapped_column( Integer, nullable=False, server_default="0" ) cases_skipped: Mapped[int] = mapped_column( Integer, nullable=False, server_default="0" ) cases_updated: Mapped[int] = mapped_column( Integer, nullable=False, server_default="0" ) errors: Mapped[Optional[str]] = mapped_column(Text, nullable=True) details: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) imported_by: Mapped[Optional[int]] = mapped_column( Integer, ForeignKey("users.id"), nullable=True ) imported_at: Mapped[dt.datetime] = mapped_column( DateTime, nullable=False, server_default=func.now() ) # Relationships imported_by_user: Mapped[Optional["User"]] = relationship( foreign_keys=[imported_by] ) __table_args__ = ( CheckConstraint( "import_type IN ('csv_crm','icd_xlsx','historical_excel','excel_sync')", name="chk_imp_type", ), ) class AuditLog(Base): __tablename__ = "audit_log" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) user_id: Mapped[Optional[int]] = mapped_column( Integer, ForeignKey("users.id"), nullable=True ) action: Mapped[str] = mapped_column(String(100), nullable=False) entity_type: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) entity_id: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) old_values: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) new_values: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) ip_address: Mapped[Optional[str]] = mapped_column(String(45), nullable=True) user_agent: Mapped[Optional[str]] = mapped_column(Text, nullable=True) created_at: Mapped[dt.datetime] = mapped_column( DateTime, nullable=False, server_default=func.now() ) # Relationships user: Mapped[Optional["User"]] = relationship(foreign_keys=[user_id]) __table_args__ = ( Index("idx_user", "user_id"), Index("idx_entity", "entity_type", "entity_id"), Index("idx_created", "created_at"), ) class Notification(Base): __tablename__ = "notifications" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) recipient_id: Mapped[int] = mapped_column( Integer, ForeignKey("users.id"), nullable=False ) notification_type: Mapped[str] = mapped_column(String(50), nullable=False) title: Mapped[str] = mapped_column(String(255), nullable=False) message: Mapped[Optional[str]] = mapped_column(Text, nullable=True) related_entity_type: Mapped[Optional[str]] = mapped_column( String(50), nullable=True ) related_entity_id: Mapped[Optional[int]] = mapped_column( Integer, nullable=True ) is_read: Mapped[bool] = mapped_column( Boolean, nullable=False, server_default="0" ) email_sent: Mapped[bool] = mapped_column( Boolean, nullable=False, server_default="0" ) email_sent_at: Mapped[Optional[dt.datetime]] = mapped_column( DateTime, nullable=True ) created_at: Mapped[dt.datetime] = mapped_column( DateTime, nullable=False, server_default=func.now() ) # Relationships recipient: Mapped["User"] = relationship(foreign_keys=[recipient_id]) __table_args__ = ( Index("idx_recipient", "recipient_id", "is_read"), CheckConstraint( "notification_type IN (" "'new_cases_uploaded','icd_entered','icd_uploaded'," "'report_ready','coding_completed')", name="chk_notif", ), )