From 048a6c6f382598fd7efaa92a5265caceb9baa6b4 Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 27 Oct 2026 18:39:53 +0100 Subject: [PATCH] Add temporal validity and lifecycle tracking to categories --- .../d9c6533f8a1b_add_category_validity.py | 102 +++++++++++++++ src/ingest/b4w.py | 117 +++++++++++++++++- src/models/b4w.py | 76 +++++++++++- tests/test_helper_db.py | 87 ++++++++++++- 4 files changed, 373 insertions(+), 9 deletions(-) create mode 100644 alembic/versions/d9c6533f8a1b_add_category_validity.py diff --git a/alembic/versions/d9c6533f8a1b_add_category_validity.py b/alembic/versions/d9c6533f8a1b_add_category_validity.py new file mode 100644 index 0000000..9ff6fd7 --- /dev/null +++ b/alembic/versions/d9c6533f8a1b_add_category_validity.py @@ -0,0 +1,102 @@ +"""add temporal validity to categories + +Revision ID: d9c6533f8a1b +Revises: 78ec55b9e1d1 +Create Date: 2026-10-25 00:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "d9c6533f8a1b" +down_revision: str | Sequence[str] | None = "78ec55b9e1d1" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + with op.batch_alter_table("category") as batch_op: + batch_op.drop_constraint("uq_category_name_parent", type_="unique") + batch_op.add_column( + sa.Column( + "valid_from", + sa.DateTime(timezone=True), + nullable=True, + info={ + "description": "UTC start of validity; NULL means valid since the indefinite past." + }, + ) + ) + batch_op.add_column( + sa.Column( + "valid_to", + sa.DateTime(timezone=True), + nullable=True, + info={ + "description": "UTC end of validity; NULL means the version is currently valid." + }, + ) + ) + batch_op.add_column( + sa.Column( + "last_action", + sa.String(), + nullable=False, + server_default="created", + info={ + "description": "Most recent lifecycle action recorded for this category version." + }, + ) + ) + + # Existing rows predate temporal tracking, so their start is unbounded. + op.execute("UPDATE category SET last_action = 'migrated'") + + # Separate root and non-root indexes so NULL parent IDs are unique too. + op.create_index( + "uq_category_active_name_parent", + "category", + ["name", "parent_id"], + unique=True, + sqlite_where=sa.text("valid_to IS NULL AND parent_id IS NOT NULL"), + postgresql_where=sa.text("valid_to IS NULL AND parent_id IS NOT NULL"), + ) + op.create_index( + "uq_category_active_root_name", + "category", + ["name"], + unique=True, + sqlite_where=sa.text("valid_to IS NULL AND parent_id IS NULL"), + postgresql_where=sa.text("valid_to IS NULL AND parent_id IS NULL"), + ) + + +def downgrade() -> None: + bind = op.get_bind() + has_temporal_history = bind.execute( + sa.text( + "SELECT 1 FROM category " + "WHERE valid_from IS NOT NULL OR valid_to IS NOT NULL LIMIT 1" + ) + ).first() + if has_temporal_history: + raise RuntimeError( + "Cannot downgrade category validity while temporal history exists; " + "the downgrade would discard it." + ) + + op.drop_index("uq_category_active_root_name", table_name="category") + op.drop_index("uq_category_active_name_parent", table_name="category") + + with op.batch_alter_table("category") as batch_op: + batch_op.drop_column("last_action") + batch_op.drop_column("valid_to") + batch_op.drop_column("valid_from") + batch_op.create_unique_constraint( + "uq_category_name_parent", ["name", "parent_id"] + ) diff --git a/src/ingest/b4w.py b/src/ingest/b4w.py index 6319958..db85e1a 100644 --- a/src/ingest/b4w.py +++ b/src/ingest/b4w.py @@ -1,4 +1,6 @@ +from datetime import UTC, datetime from pathlib import Path +from typing import Any from sqlalchemy.orm import Session @@ -17,20 +19,125 @@ def upsert_category( ) -> CategoryDb | None: if category is None: return None - parameters = { - "name": category.name, - "parent": upsert_category(session, category.parent), + parent = upsert_category(session, category.parent, do_commit=do_commit) + parent_id = parent.id if parent else None + matching_rows = session.query(CategoryDb).filter_by( + name=category.name, parent_id=parent_id + ) + active = matching_rows.filter(CategoryDb.valid_to.is_(None)).one_or_none() + if active is not None: + return active + + parameters = {"name": category.name, "parent": parent} + parameters_existing = {**parameters, "valid_to": None} + parameters_new: dict[str, Any] = { + **parameters, + "valid_from": None, + "valid_to": None, + "last_action": "created", } + historical = matching_rows.filter(CategoryDb.valid_to.is_not(None)).first() + if historical is not None: + parameters_new["valid_from"] = _utc_datetime(None) + parameters_new["last_action"] = "changed" + return _upsert( session, CategoryDb, - parameters, - parameters, + parameters_existing, + parameters_new, do_update=False, do_commit=do_commit, ) +def _utc_datetime(value: datetime | None) -> datetime: + if value is None: + return datetime.now(UTC) + if value.tzinfo is None or value.utcoffset() is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def _ensure_after_valid_from(category: CategoryDb, effective_at: datetime) -> None: + if category.valid_from is None: + return + valid_from = _utc_datetime(category.valid_from) + if effective_at <= valid_from: + raise ValueError("The action time must be later than valid_from") + + +def change_category( + session: Session, + category_id: int, + replacement: Category, + effective_at: datetime | None = None, + do_commit: bool = True, +) -> CategoryDb: + """Close an active category version and start its replacement version.""" + current = ( + session.query(CategoryDb).filter_by(id=category_id, valid_to=None).one_or_none() + ) + if current is None: + raise ValueError(f"No active category found with id {category_id}") + + action_time = _utc_datetime(effective_at) + _ensure_after_valid_from(current, action_time) + parent = upsert_category(session, replacement.parent, do_commit=False) + parent_id = parent.id if parent else None + + if current.name == replacement.name and current.parent_id == parent_id: + return current + + replacement_active = ( + session.query(CategoryDb) + .filter_by(name=replacement.name, parent_id=parent_id, valid_to=None) + .one_or_none() + ) + + current.valid_to = action_time + current.last_action = "changed" + session.flush() + + if replacement_active is None: + replacement_active = CategoryDb( + name=replacement.name, + parent=parent, + valid_from=action_time, + valid_to=None, + last_action="changed", + ) + session.add(replacement_active) + session.flush() + + if do_commit: + session.commit() + return replacement_active + + +def retire_category( + session: Session, + category_id: int, + effective_at: datetime | None = None, + do_commit: bool = True, +) -> CategoryDb: + """Close an active category version without creating a replacement.""" + category = ( + session.query(CategoryDb).filter_by(id=category_id, valid_to=None).one_or_none() + ) + if category is None: + raise ValueError(f"No active category found with id {category_id}") + + action_time = _utc_datetime(effective_at) + _ensure_after_valid_from(category, action_time) + category.valid_to = action_time + category.last_action = "retired" + session.flush() + if do_commit: + session.commit() + return category + + def upsert_b4w( session: Session, data: B4wExport, do_update=False, do_commit=True ) -> B4wDb | None: diff --git a/src/models/b4w.py b/src/models/b4w.py index 364d727..d9dcf00 100644 --- a/src/models/b4w.py +++ b/src/models/b4w.py @@ -1,8 +1,17 @@ -from datetime import date +from datetime import UTC, date, datetime from decimal import Decimal from pydantic import Field, computed_field, field_validator -from sqlalchemy import Date, ForeignKey, Integer, String, UniqueConstraint +from sqlalchemy import ( + Date, + DateTime, + ForeignKey, + Index, + Integer, + String, + TypeDecorator, + text, +) from sqlalchemy.orm import Mapped, mapped_column, relationship from src.constants import IBAN_PATTERN, Currency @@ -118,10 +127,48 @@ class Category(ExportBaseModel): return f'Category.from_str("{self.__str__()}")' +class UTCDateTime(TypeDecorator[datetime]): + """Store UTC timestamps and return timezone-aware UTC values on every dialect.""" + + impl = DateTime + cache_ok = True + + def load_dialect_impl(self, dialect): + return dialect.type_descriptor(DateTime(timezone=True)) + + def process_bind_param(self, value: datetime | None, dialect): + if value is None: + return None + if value.tzinfo is None or value.utcoffset() is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + def process_result_value(self, value: datetime | None, dialect): + if value is None: + return None + if value.tzinfo is None or value.utcoffset() is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + class CategoryDb(BaseModelDb): __tablename__ = "category" __table_args__ = ( - UniqueConstraint("name", "parent_id", name="uq_category_name_parent"), + Index( + "uq_category_active_name_parent", + "name", + "parent_id", + unique=True, + sqlite_where=text("valid_to IS NULL AND parent_id IS NOT NULL"), + postgresql_where=text("valid_to IS NULL AND parent_id IS NOT NULL"), + ), + Index( + "uq_category_active_root_name", + "name", + unique=True, + sqlite_where=text("valid_to IS NULL AND parent_id IS NULL"), + postgresql_where=text("valid_to IS NULL AND parent_id IS NULL"), + ), ) id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) @@ -129,6 +176,29 @@ class CategoryDb(BaseModelDb): parent_id: Mapped[int | None] = mapped_column( ForeignKey("category.id"), nullable=True ) + valid_from: Mapped[datetime | None] = mapped_column( + UTCDateTime(), + nullable=True, + info={ + "description": "UTC start of validity; NULL means valid since the indefinite past." + }, + ) + valid_to: Mapped[datetime | None] = mapped_column( + UTCDateTime(), + nullable=True, + info={ + "description": "UTC end of validity; NULL means the version is currently valid." + }, + ) + last_action: Mapped[str] = mapped_column( + String, + nullable=False, + default="created", + server_default="created", + info={ + "description": "Most recent lifecycle action recorded for this category version." + }, + ) # Relationships parent = relationship( diff --git a/tests/test_helper_db.py b/tests/test_helper_db.py index a25f97d..e1cd4c1 100644 --- a/tests/test_helper_db.py +++ b/tests/test_helper_db.py @@ -1,7 +1,11 @@ +from datetime import UTC, datetime + +import pytest from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session -from src.ingest.b4w import upsert_category +from src.ingest.b4w import change_category, retire_category, upsert_category from src.models import Category from src.models.b4w import CategoryDb @@ -17,6 +21,13 @@ def test_upsert_category_returns_orm_object(db_session: Session): assert str(result.name) == "dummy" assert str(result.parent.name) == "dummy_parent_2" assert str(result.parent.parent.name) == "dummy_parent_1" + assert result.valid_from is None + assert result.valid_to is None + assert result.last_action == "created" + assert all( + CategoryDb.__table__.c[name].info.get("description") + for name in ("valid_from", "valid_to", "last_action") + ) def test_upsert_category_writes_new_to_db(db_session: Session): @@ -45,3 +56,77 @@ def test_upsert_category_skips_existing(db_session: Session): select(CategoryDb).filter_by(name="try_again_and_again") ).all() assert len(result) == 1 + + +def test_change_category_closes_old_version_and_starts_new_one( + db_session: Session, +): + original = upsert_category( + db_session, Category.from_str("old_category"), do_commit=False + ) + assert original is not None + effective_at = datetime(2026, 10, 16, 12, tzinfo=UTC) + + replacement = change_category( + db_session, + original.id, + Category.from_str("new_category"), + effective_at=effective_at, + do_commit=False, + ) + + assert original.valid_from is None + assert original.valid_to == effective_at + assert original.last_action == "changed" + assert replacement.name == "new_category" + assert replacement.valid_from == effective_at + assert replacement.valid_to is None + assert replacement.last_action == "changed" + + db_session.expire(original) + db_session.expire(replacement) + assert original.valid_to == effective_at + assert replacement.valid_from == effective_at + + # A historical name can be introduced again as a new active version. + reintroduced = upsert_category( + db_session, Category.from_str("old_category"), do_commit=False + ) + assert reintroduced is not None + assert reintroduced.id != original.id + assert reintroduced.valid_from is not None + assert reintroduced.valid_from > original.valid_to + assert reintroduced.last_action == "changed" + + +def test_retire_category_closes_active_version(db_session: Session): + category = upsert_category( + db_session, Category.from_str("category_to_retire"), do_commit=False + ) + assert category is not None + effective_at = datetime(2026, 10, 16, 12, tzinfo=UTC) + + retired = retire_category( + db_session, category.id, effective_at=effective_at, do_commit=False + ) + + assert retired.valid_to == effective_at + assert retired.last_action == "retired" + assert ( + db_session.query(CategoryDb) + .filter_by(name="category_to_retire", valid_to=None) + .one_or_none() + is None + ) + + +def test_active_root_categories_are_unique(db_session: Session): + db_session.add_all( + [ + CategoryDb(name="duplicate_root"), + CategoryDb(name="duplicate_root"), + ] + ) + + with pytest.raises(IntegrityError): + db_session.flush()