src/helper/__init__.py

@@ -1 +0,0 @@
-"""Database helper for application-specific persistence operations."""

src/helper/category.py

@@ -1,22 +0,0 @@
-from sqlalchemy.orm import Session
-
-from src.db_utils import upsert
-from src.models.category import Category, CategoryDb
-
-
-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,
-    )

src/ingest/b4w.py

@@ -3,10 +3,10 @@ from pathlib import Path
 from sqlalchemy.orm import Session
 
 from src.db_utils import upsert
-from src.helper.category import upsert_category
 from src.ingest.common import _reader_raw
 from src.ingest.transactions import consolidate_transactions, map_b4w_to_transaction
 from src.models.b4w import B4wDb, B4wExport
+from src.models.category import upsert_category
 
 
 def reader_b4w(path: Path) -> list[B4wExport]:

src/models/category.py

@@ -1,7 +1,9 @@
 from pydantic import Field, computed_field
 from sqlalchemy import ForeignKey, String, UniqueConstraint
-from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.orm import Mapped, Session, mapped_column, relationship
 
+from src.db_utils import upsert
+from src.models.category_budget import CategoryBudgetDb
 from src.models.common import BaseModelDb, ExportBaseModel, hashify
 
 
@@ -53,4 +55,22 @@ class CategoryDb(BaseModelDb):
         "CategoryDb", remote_side="CategoryDb.id", back_populates="children"
     )
     children = relationship("CategoryDb", back_populates="parent")
-    budgets = relationship("CategoryBudgetDb", back_populates="category")
+    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,
+    )

tests/test_helper_db.py

@@ -1,47 +0,0 @@
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from src.helper.category import upsert_category
-from src.models import Category
-from src.models.category import CategoryDb
-
-
-def test_upsert_category_returns_orm_object(db_session: Session):
-    result = upsert_category(
-        db_session,
-        Category.from_str("dummy_parent_1:dummy_parent_2:dummy"),
-        do_commit=False,
-    )
-    assert result
-    assert isinstance(result, CategoryDb)
-    assert str(result.name) == "dummy"
-    assert str(result.parent.name) == "dummy_parent_2"
-    assert str(result.parent.parent.name) == "dummy_parent_1"
-
-
-def test_upsert_category_writes_new_to_db(db_session: Session):
-    upsert_category(
-        db_session,
-        Category.from_str("grand_father:father:new_category"),
-        do_commit=True,
-    )
-    result = db_session.execute(
-        select(CategoryDb).filter_by(name="new_category")
-    ).scalar_one()
-    assert result
-    assert str(result.name) == "new_category"
-    assert str(result.parent.name) == "father"
-    assert str(result.parent.parent.name) == "grand_father"
-
-
-def test_upsert_category_skips_existing(db_session: Session):
-    for _ in range(3):
-        upsert_category(
-            db_session,
-            Category.from_str("grand_father:father:try_again_and_again"),
-            do_commit=True,
-        )
-    result = db_session.execute(
-        select(CategoryDb).filter_by(name="try_again_and_again")
-    ).all()
-    assert len(result) == 1

tests/test_model_category.py

@@ -1,56 +1,45 @@
-from src.models import Category
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from src.models.category import Category, CategoryDb, upsert_category
 
 
-def test_single_level_category():
+def test_upsert_category_returns_orm_object(db_session: Session):
-    c = Category.from_str("single")
+    result = upsert_category(
-    assert c
+        db_session,
-    assert c.name == "single"
+        Category.from_str("dummy_parent_1:dummy_parent_2:dummy"),
-    assert c.parent is None
+        do_commit=False,
-
-
-def test_two_level_category():
-    c = Category.from_str("first:second")
-    assert c
-    assert c.name == "second"
-    assert isinstance(c.parent, Category)
-    assert c.parent.name == "first"
-    assert c.parent.parent is None
-
-
-def test_three_level_category():
-    c = Category.from_str("first:second:third")
-    assert c
-    assert c.name == "third"
-    assert isinstance(c.parent, Category)
-    assert c.parent.name == "second"
-    assert isinstance(c.parent.parent, Category)
-    assert c.parent.parent.name == "first"
-    assert c.parent.parent.parent is None
-
-
-def test_single_level_category_to_string():
-    c = Category.from_str("single")
-    assert c
-    assert str(c) == "single"
-
-
-def test_two_level_category_to_string():
-    c = Category(name="second", parent=Category(name="first"))
-    assert c.to_str() == "first:second"
-
-
-def test_three_level_category_to_string():
-    c = Category(
-        name="third", parent=Category(name="second", parent=Category(name="first"))
     )
-    assert c.to_str() == "first:second:third"
+    assert result
+    assert isinstance(result, CategoryDb)
+    assert str(result.name) == "dummy"
+    assert str(result.parent.name) == "dummy_parent_2"
+    assert str(result.parent.parent.name) == "dummy_parent_1"
 
 
-def test_category_str():
+def test_upsert_category_writes_new_to_db(db_session: Session):
-    c = Category.from_str("first:second:third")
+    upsert_category(
-    assert str(c) == "first:second:third"
+        db_session,
+        Category.from_str("grand_father:father:new_category"),
+        do_commit=True,
+    )
+    result = db_session.execute(
+        select(CategoryDb).filter_by(name="new_category")
+    ).scalar_one()
+    assert result
+    assert str(result.name) == "new_category"
+    assert str(result.parent.name) == "father"
+    assert str(result.parent.parent.name) == "grand_father"
 
 
-def test_category_repr():
+def test_upsert_category_skips_existing(db_session: Session):
-    c = Category.from_str("first:second:third")
+    for _ in range(3):
-    assert repr(c) == 'Category.from_str("first:second:third")'
+        upsert_category(
+            db_session,
+            Category.from_str("grand_father:father:try_again_and_again"),
+            do_commit=True,
+        )
+    result = db_session.execute(
+        select(CategoryDb).filter_by(name="try_again_and_again")
+    ).all()
+    assert len(result) == 1