Extract Category model into src/models/category.py
This commit is contained in:
+2
-1
@@ -4,7 +4,8 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from src.ingest._common import _reader_raw, _upsert
|
||||
from src.ingest.transactions import consolidate_transactions, map_b4w_to_transaction
|
||||
from src.models.b4w import B4wDb, B4wExport, Category, CategoryDb
|
||||
from src.models.b4w import B4wDb, B4wExport, Category
|
||||
from src.models.category import CategoryDb
|
||||
|
||||
|
||||
def reader_b4w(path: Path) -> list[B4wExport]:
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
from src.models.b4w import B4wExport as B4wExport
|
||||
from src.models.b4w import Category as Category
|
||||
from src.models.category import Category as Category
|
||||
from src.models.category_budget import CategoryBudgetDb as CategoryBudgetDb
|
||||
from src.models.comdirect import ComdirectCreditCard as ComdirectCreditCard
|
||||
from src.models.comdirect import ComdirectDepot as ComdirectDepot
|
||||
|
||||
+2
-53
@@ -2,18 +2,18 @@ from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from pydantic import Field, computed_field, field_validator
|
||||
from sqlalchemy import Date, ForeignKey, Integer, String, UniqueConstraint
|
||||
from sqlalchemy import Date, ForeignKey, Integer, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from src.constants import IBAN_PATTERN, Currency
|
||||
from src.models._common import (
|
||||
BaseModelDb,
|
||||
ExportBaseModel,
|
||||
IngestdbBase,
|
||||
Sign,
|
||||
_hash,
|
||||
date2str,
|
||||
)
|
||||
from src.models.category import Category
|
||||
|
||||
|
||||
class B4wExport(ExportBaseModel):
|
||||
@@ -87,57 +87,6 @@ class B4wExport(ExportBaseModel):
|
||||
return category
|
||||
|
||||
|
||||
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 _hash(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 = relationship("CategoryBudgetDb", back_populates="category")
|
||||
|
||||
|
||||
class B4wDb(IngestdbBase):
|
||||
__tablename__ = "import_b4w"
|
||||
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
from pydantic import Field, computed_field
|
||||
from sqlalchemy import ForeignKey, String, UniqueConstraint
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from src.models._common import BaseModelDb, ExportBaseModel, _hash
|
||||
|
||||
|
||||
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 _hash(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 = relationship("CategoryBudgetDb", back_populates="category")
|
||||
Reference in New Issue
Block a user