from datetime import date from decimal import Decimal from typing import Self from pydantic import BaseModel, Field, computed_field, model_validator from sqlalchemy import Date, ForeignKey, Numeric, String, UniqueConstraint from sqlalchemy.orm import Mapped, Session, mapped_column, relationship from src.db_utils import upsert from src.models.common import BaseModelDb, ExportBaseModel, hashify class Category(ExportBaseModel): name: str = Field(pattern=r"[^:\r\n]+") parent: Category | None = None @computed_field # type: ignore[prop-decorator] @property def hash_id(self) -> str: return hashify(self.name, self.parent) @classmethod def from_str(cls, s: str) -> Category | None: if not s: return None components = s.split(":") name = components[-1] rest = ":".join(components[:-1]) return cls(name=name, parent=cls.from_str(rest)) def to_str(self) -> str | None: if not self.parent: return self.name else: return f"{self.parent.to_str()}:{self.name}" def __str__(self): return self.to_str() def __repr__(self): return f'Category.from_str("{self.__str__()}")' class CategoryDb(BaseModelDb): __tablename__ = "category" __table_args__ = ( UniqueConstraint("name", "parent_id", name="uq_category_name_parent"), ) id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String, nullable=False) parent_id: Mapped[int | None] = mapped_column( ForeignKey("category.id"), nullable=True ) # Relationships parent = relationship( "CategoryDb", remote_side="CategoryDb.id", back_populates="children" ) children = relationship("CategoryDb", back_populates="parent") budgets: Mapped[list[CategoryBudgetDb]] = relationship(back_populates="category") def upsert_category( session: Session, category: Category | None, do_commit: bool = True ) -> CategoryDb | None: if category is None: return None parent = upsert_category(session, category.parent, do_commit=False) parameters = {"name": category.name, "parent": parent} return upsert( session, CategoryDb, parameters, parameters, do_update=False, do_commit=do_commit, ) class CategoryBudget(BaseModel): category: Category amount_eur: Decimal = Field(gt=0, decimal_places=2) valid_from: date | None = Field( default=None, description="Start of validity. None means unbounded" ) valid_to: date | None = Field( default=None, description="End of validity. None means unbounded" ) @model_validator(mode="after") def validate_date_range(self) -> Self: if ( self.valid_from is not None and self.valid_to is not None and self.valid_from > self.valid_to ): raise ValueError( f"`valid_from` ({self.valid_from}) and valid_to ({self.valid_to}) are nor in order" ) return self class CategoryBudgetDb(BaseModelDb): """Budget amount for a category, optionally bounded to a time window. Satellite table to ``category``: one category may have several budgets (e.g. one per month). ``valid_from``/``valid_to`` are nullable; ``None`` means unbounded on that side. """ __tablename__ = "category_budget" id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) category_id: Mapped[int] = mapped_column(ForeignKey("category.id"), nullable=False) amount_eur: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False) valid_from: Mapped[date | None] = mapped_column(Date, nullable=True, default=None) valid_to: Mapped[date | None] = mapped_column(Date, nullable=True, default=None) # relationship category = relationship("CategoryDb", back_populates="budgets")