import enum from datetime import date, datetime from decimal import Decimal from typing import Annotated, Optional from pydantic import PlainSerializer from sqlalchemy import JSON, Column, UniqueConstraint from sqlalchemy.dialects.postgresql import JSONB from sqlmodel import Field, Relationship, SQLModel from app.timeutil import utcnow # Live DBs store usersettings.data as jsonb; keep plain JSON elsewhere and on # SQLite (tests). JSON_OR_JSONB = JSON().with_variant(JSONB(), "postgresql") # Monetary values: exact NUMERIC storage, but serialized to JSON as plain # numbers so the API contract (frontend expects `number`) is unchanged. # Pinned by tests/test_models_serialization.py. Money = Annotated[ Decimal, PlainSerializer(float, return_type=float, when_used="json") ] class RecurringItemType(str, enum.Enum): INCOME = "INCOME" EXPENSE = "EXPENSE" SAVINGS = "SAVINGS" class RecurringFrequency(str, enum.Enum): WEEKLY = "WEEKLY" MONTHLY = "MONTHLY" QUARTERLY = "QUARTERLY" BIANNUAL = "BIANNUAL" YEARLY = "YEARLY" class TransactionType(str, enum.Enum): COMPRA = "COMPRA" DEVOLUCION = "DEVOLUCION" DEPOSITO = "DEPOSITO" SALARY = "SALARY" class TransactionSource(str, enum.Enum): CREDIT_CARD = "CREDIT_CARD" CASH = "CASH" TRANSFER = "TRANSFER" class Currency(str, enum.Enum): CRC = "CRC" USD = "USD" EUR = "EUR" BTC = "BTC" XMR = "XMR" class Bank(str, enum.Enum): BAC = "BAC" BCR = "BCR" DAVIVIENDA = "DAVIVIENDA" FCL = "FCL" ROP = "ROP" VOL = "VOL" MEMP = "MEMP" MPAT = "MPAT" MORTGAGE = "MORTGAGE" class AccountType(str, enum.Enum): BANK = "BANK" PENSION = "PENSION" CRYPTO = "CRYPTO" SAVINGS = "SAVINGS" LIABILITY = "LIABILITY" # --- Category --- class CategoryBase(SQLModel): name: str = Field(index=True, unique=True) icon: str = "tag" auto_match_patterns: Optional[str] = Field( default=None, description="Comma-separated merchant substrings for auto-matching", ) class Category(CategoryBase, table=True): id: Optional[int] = Field(default=None, primary_key=True) transactions: list["Transaction"] = Relationship(back_populates="category") class CategoryCreate(CategoryBase): pass class CategoryRead(CategoryBase): id: int class CategoryUpdate(SQLModel): name: Optional[str] = None icon: Optional[str] = None auto_match_patterns: Optional[str] = None # --- Account --- class AccountBase(SQLModel): bank: Bank currency: Currency label: str balance: Money = Field(default=Decimal("0"), max_digits=15, decimal_places=2) account_type: AccountType = AccountType.BANK next_payment: Optional[Money] = Field(default=None, max_digits=15, decimal_places=2) class Account(AccountBase, table=True): id: Optional[int] = Field(default=None, primary_key=True) updated_at: datetime = Field(default_factory=utcnow) class AccountCreate(AccountBase): pass class AccountRead(AccountBase): id: int updated_at: datetime class AccountUpdate(SQLModel): balance: Optional[Money] = None label: Optional[str] = None next_payment: Optional[Money] = None # --- Transaction --- class TransactionBase(SQLModel): amount: Money = Field(max_digits=15, decimal_places=2) currency: Currency = Currency.CRC merchant: str city: Optional[str] = None date: datetime card_type: Optional[str] = None card_last4: Optional[str] = None authorization_code: Optional[str] = None reference: Optional[str] = Field(default=None, index=True) transaction_type: TransactionType = TransactionType.COMPRA source: TransactionSource = TransactionSource.CREDIT_CARD bank: Bank = Bank.BAC notes: Optional[str] = None category_id: Optional[int] = Field(default=None, foreign_key="category.id") deferred_to_next_cycle: bool = Field( default=False, sa_column_kwargs={"server_default": "false"} ) class Transaction(TransactionBase, table=True): id: Optional[int] = Field(default=None, primary_key=True) created_at: datetime = Field(default_factory=utcnow) category: Optional[Category] = Relationship(back_populates="transactions") class TransactionCreate(TransactionBase): pass class TransactionRead(TransactionBase): id: int created_at: datetime category: Optional[CategoryRead] = None class TransactionUpdate(SQLModel): amount: Optional[Money] = None currency: Optional[Currency] = None merchant: Optional[str] = None city: Optional[str] = None date: Optional[datetime] = None transaction_type: Optional[TransactionType] = None source: Optional[TransactionSource] = None notes: Optional[str] = None category_id: Optional[int] = None deferred_to_next_cycle: Optional[bool] = None # --- Exchange Rate --- class ExchangeRate(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) date: datetime buy_rate: Decimal = Field(max_digits=15, decimal_places=6) sell_rate: Decimal = Field(max_digits=15, decimal_places=6) fetched_at: datetime = Field(default_factory=utcnow) class ExchangeRateRead(SQLModel): buy_rate: Money sell_rate: Money date: datetime fetched_at: datetime # --- API Token --- class APIToken(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) name: str token_hash: str = Field(index=True) created_at: datetime = Field(default_factory=utcnow) expires_at: Optional[datetime] = None is_active: bool = True class APITokenCreate(SQLModel): name: str expires_days: Optional[int] = None class APITokenRead(SQLModel): id: int name: str created_at: datetime expires_at: Optional[datetime] is_active: bool # --- User Settings --- class UserSettings(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) key: str = Field(index=True, unique=True, default="default") data: dict = Field( default_factory=dict, sa_column=Column(JSON_OR_JSONB, nullable=False, server_default="{}"), ) updated_at: datetime = Field(default_factory=utcnow) class UserSettingsRead(SQLModel): key: str data: dict updated_at: datetime class UserSettingsUpdate(SQLModel): data: dict # --- Recurring Item --- class RecurringItemBase(SQLModel): name: str amount: Money = Field(max_digits=15, decimal_places=2) currency: Currency = Currency.CRC item_type: RecurringItemType frequency: RecurringFrequency = RecurringFrequency.MONTHLY day_of_month: Optional[int] = None month_of_year: Optional[int] = None override_amounts: Optional[dict] = Field( default=None, sa_column=Column(JSON, nullable=True), ) category_id: Optional[int] = Field(default=None, foreign_key="category.id") is_active: bool = True notes: Optional[str] = None class RecurringItem(RecurringItemBase, table=True): id: Optional[int] = Field(default=None, primary_key=True) created_at: datetime = Field(default_factory=utcnow) category: Optional[Category] = Relationship() class RecurringItemCreate(RecurringItemBase): pass class RecurringItemRead(RecurringItemBase): id: int created_at: datetime category: Optional[CategoryRead] = None class RecurringItemUpdate(SQLModel): name: Optional[str] = None amount: Optional[Money] = None currency: Optional[Currency] = None item_type: Optional[RecurringItemType] = None frequency: Optional[RecurringFrequency] = None day_of_month: Optional[int] = None month_of_year: Optional[int] = None override_amounts: Optional[dict] = None category_id: Optional[int] = None is_active: Optional[bool] = None notes: Optional[str] = None # --- Push Subscription --- class PushSubscription(SQLModel, table=True): id: Optional[int] = Field(default=None, primary_key=True) endpoint: str = Field(unique=True) p256dh: str auth: str created_at: datetime = Field(default_factory=utcnow) class PushSubscriptionCreate(SQLModel): endpoint: str keys: dict # {"p256dh": "...", "auth": "..."} # --- Pension Snapshot --- class PensionSnapshotBase(SQLModel): fund: Bank contract_number: str period_start: date period_end: date saldo_anterior: Money = Field(max_digits=15, decimal_places=2) aportes: Money = Field(max_digits=15, decimal_places=2) rendimientos: Money = Field(max_digits=15, decimal_places=2) retiros: Money = Field(max_digits=15, decimal_places=2) traslados: Money = Field(max_digits=15, decimal_places=2) comision: Money = Field(max_digits=15, decimal_places=2) correccion: Money = Field(max_digits=15, decimal_places=2) bonificacion: Money = Field(max_digits=15, decimal_places=2) saldo_final: Money = Field(max_digits=15, decimal_places=2) source_filename: str class PensionSnapshot(PensionSnapshotBase, table=True): __table_args__ = ( UniqueConstraint("fund", "period_start", "period_end"), ) id: Optional[int] = Field(default=None, primary_key=True) created_at: datetime = Field(default_factory=utcnow) class PensionSnapshotRead(PensionSnapshotBase): id: int created_at: datetime # --- Balance Override --- class BalanceOverride(SQLModel, table=True): __table_args__ = (UniqueConstraint("year", "month"),) id: Optional[int] = Field(default=None, primary_key=True) year: int month: int override_balance: Money = Field(max_digits=15, decimal_places=2) created_at: datetime = Field(default_factory=utcnow) updated_at: datetime = Field(default_factory=utcnow) class BalanceOverrideCreate(SQLModel): override_balance: Money class BalanceOverrideRead(SQLModel): id: int year: int month: int override_balance: Money updated_at: datetime # --- Savings Accrual --- class SavingsAccrualBase(SQLModel): year: int month: int memp_amount: Money = Field(default=Decimal("200000"), max_digits=15, decimal_places=2) mpat_amount: Money = Field(default=Decimal("200000"), max_digits=15, decimal_places=2) trigger_transaction_id: Optional[int] = None notes: Optional[str] = None class SavingsAccrual(SavingsAccrualBase, table=True): __table_args__ = (UniqueConstraint("year", "month"),) id: Optional[int] = Field(default=None, primary_key=True) applied_at: datetime = Field(default_factory=utcnow) class SavingsAccrualCreate(SavingsAccrualBase): pass class SavingsAccrualRead(SavingsAccrualBase): id: int applied_at: datetime class SavingsAccrualUpdate(SQLModel): memp_amount: Optional[Money] = None mpat_amount: Optional[Money] = None notes: Optional[str] = None # --- Municipal Receipt --- class MunicipalReceiptBase(SQLModel): receipt_date: date due_date: date period: str # "YYYY-MM" account: str finca: str holder_name: str holder_cedula: str holder_address: str subtotal: Money = Field(max_digits=15, decimal_places=2) interests: Money = Field(max_digits=15, decimal_places=2) iva: Money = Field(max_digits=15, decimal_places=2) total: Money = Field(max_digits=15, decimal_places=2) raw_charges: list[dict] = Field( default_factory=list, sa_column=Column(JSON, nullable=False, server_default="[]"), ) source_filename: str class MunicipalReceipt(MunicipalReceiptBase, table=True): __table_args__ = (UniqueConstraint("account", "period"),) id: Optional[int] = Field(default=None, primary_key=True) created_at: datetime = Field(default_factory=utcnow) water_readings: list["WaterMeterReading"] = Relationship( back_populates="receipt", ) class MunicipalReceiptCreate(MunicipalReceiptBase): pass class MunicipalReceiptRead(MunicipalReceiptBase): id: int created_at: datetime # --- Water Meter Reading --- class WaterMeterReadingBase(SQLModel): meter_id: str period: str # "YYYY-MM" reading_previous: Money = Field(default=Decimal("0"), max_digits=15, decimal_places=2) reading_current: Money = Field(default=Decimal("0"), max_digits=15, decimal_places=2) consumption_m3: Money = Field(max_digits=15, decimal_places=2) agua_potable: Money = Field(default=Decimal("0"), max_digits=15, decimal_places=2) serv_ambientales: Money = Field(default=Decimal("0"), max_digits=15, decimal_places=2) alcant_sanitario: Money = Field(default=Decimal("0"), max_digits=15, decimal_places=2) iva: Money = Field(default=Decimal("0"), max_digits=15, decimal_places=2) is_historical: bool = False receipt_id: Optional[int] = Field(default=None, foreign_key="municipalreceipt.id") class WaterMeterReading(WaterMeterReadingBase, table=True): __table_args__ = (UniqueConstraint("meter_id", "period", "is_historical"),) id: Optional[int] = Field(default=None, primary_key=True) created_at: datetime = Field(default_factory=utcnow) receipt: Optional[MunicipalReceipt] = Relationship( back_populates="water_readings", ) class WaterMeterReadingRead(WaterMeterReadingBase): id: int created_at: datetime