""" app/models.py ───────────── SQLAlchemy ORM models (database schema) and Pydantic v2 schemas (request/response validation) for every PMDI domain entity. """ from __future__ import annotations from datetime import date, datetime from typing import Any, Dict, List, Optional from pydantic import BaseModel, ConfigDict, EmailStr from sqlalchemy import ( Boolean, Column, Date, DateTime, ForeignKey, Integer, String, Text, ) from sqlalchemy.dialects.postgresql import ARRAY, JSONB from sqlalchemy.orm import relationship from app.database import Base # ═══════════════════════════════════════════════════════════════════════════════ # Notification delivery-status constants # ═══════════════════════════════════════════════════════════════════════════════ STATUS_DELIVERED = "Delivered from Server" STATUS_RELAY = "Sent to Mail Relay" STATUS_FAILED = "Failed" # ═══════════════════════════════════════════════════════════════════════════════ # SQLAlchemy ORM Models # ═══════════════════════════════════════════════════════════════════════════════ class AdminUser(Base): """Internal administrator account with optional TOTP MFA.""" __tablename__ = "admin_users" id = Column(Integer, primary_key=True, index=True) username = Column(String(64), unique=True, nullable=False, index=True) hashed_password = Column(String(256), nullable=False) totp_secret = Column(String(64), nullable=True) # base32 TOTP secret mfa_enabled = Column(Boolean, default=False, nullable=False) created_at = Column(DateTime, default=datetime.utcnow, nullable=False) class Client(Base): """Customer record referenced by invoices.""" __tablename__ = "clients" id = Column(Integer, primary_key=True, index=True) client_id_name = Column(String(64), unique=True, nullable=False) # short slug client_name = Column(String(256), nullable=False) client_email = Column(String(256), nullable=False) created_at = Column(DateTime, default=datetime.utcnow) invoices = relationship("Invoice", back_populates="client") class EmailTemplate(Base): """Jinja2 email template with a human-readable name, subject, and body.""" __tablename__ = "email_templates" id = Column(Integer, primary_key=True, index=True) name = Column(String(128), nullable=False) # dropdown label template_type = Column(String(32), nullable=False) # "Notification" | "Overdue" subject = Column(String(256), nullable=False) # Jinja2 subject line template_body = Column(Text, nullable=False) # Jinja2 body created_at = Column(DateTime, default=datetime.utcnow) invoices = relationship("Invoice", back_populates="template") class Invoice(Base): """Open receivable with scheduling metadata for automated reminders.""" __tablename__ = "invoices" id = Column(Integer, primary_key=True, index=True) invoice_number = Column(String(64), nullable=False, unique=True) client_id = Column(Integer, ForeignKey("clients.id", ondelete="RESTRICT"), nullable=False) pay_date = Column(Date, nullable=False) template_id = Column(Integer, ForeignKey("email_templates.id", ondelete="RESTRICT"), nullable=False) # Days-before-pay_date triggers, e.g. [30, 14, 7, 1] notification_dates = Column(ARRAY(Integer), nullable=True, default=[]) # Send recurring overdue alert every N days past due (0 = disabled) overdue_recurring_days = Column(Integer, default=0, nullable=False) paid = Column(Boolean, default=False, nullable=False) created_at = Column(DateTime, default=datetime.utcnow) client = relationship("Client", back_populates="invoices") template = relationship("EmailTemplate", back_populates="invoices") logs = relationship( "NotificationLog", back_populates="invoice", cascade="all, delete-orphan", ) class NotificationLog(Base): """Immutable audit record for every email dispatch attempt.""" __tablename__ = "notification_logs" id = Column(Integer, primary_key=True, index=True) invoice_id = Column(Integer, ForeignKey("invoices.id", ondelete="CASCADE"), nullable=False) status = Column(String(64), nullable=False) # STATUS_* constants sent_at = Column(DateTime, default=datetime.utcnow, nullable=False) context = Column(JSONB, nullable=True) # runtime snapshot for UI display error_message = Column(Text, nullable=True) invoice = relationship("Invoice", back_populates="logs") # ═══════════════════════════════════════════════════════════════════════════════ # Pydantic v2 Schemas # ═══════════════════════════════════════════════════════════════════════════════ _orm = ConfigDict(from_attributes=True) # ── Auth / JWT ───────────────────────────────────────────────────────────────── class LoginRequest(BaseModel): username: str password: str class TokenResponse(BaseModel): access_token: str token_type: str = "bearer" mfa_required: bool = False class MFAVerifyRequest(BaseModel): username: str totp_code: str class MFASetupResponse(BaseModel): secret: str qr_uri: str class UserRead(BaseModel): model_config = _orm id: int username: str mfa_enabled: bool # ── Client ───────────────────────────────────────────────────────────────────── class ClientCreate(BaseModel): client_id_name: str client_name: str client_email: EmailStr class ClientUpdate(ClientCreate): pass class ClientRead(BaseModel): model_config = _orm id: int client_id_name: str client_name: str client_email: str # ── EmailTemplate ────────────────────────────────────────────────────────────── class TemplateCreate(BaseModel): name: str template_type: str # "Notification" | "Overdue" subject: str template_body: str class TemplateUpdate(TemplateCreate): pass class TemplateRead(BaseModel): model_config = _orm id: int name: str template_type: str subject: str template_body: str # ── Invoice ──────────────────────────────────────────────────────────────────── class InvoiceCreate(BaseModel): invoice_number: str client_id: int pay_date: date template_id: int notification_dates: Optional[List[int]] = [] overdue_recurring_days: int = 0 paid: bool = False class InvoiceUpdate(InvoiceCreate): pass class InvoiceRead(BaseModel): model_config = _orm id: int invoice_number: str client_id: int pay_date: date template_id: int notification_dates: Optional[List[int]] = [] overdue_recurring_days: int paid: bool # Eagerly joined for the dashboard display client: Optional[ClientRead] = None template: Optional[TemplateRead] = None # ── Notification Log ─────────────────────────────────────────────────────────── class NotificationLogRead(BaseModel): model_config = _orm id: int invoice_id: int status: str sent_at: datetime context: Optional[Dict[str, Any]] = None error_message: Optional[str] = None # ── Config ───────────────────────────────────────────────────────────────────── class ConfigEntry(BaseModel): key: str value: str masked: bool