Compare commits

3 Commits
Author SHA1 Message Date
mathias2784 6babc560c7 Category overhaul (#20)
Tests / test (push) Successful in 16s
Tests / lint (push) Successful in 16s
Reviewed-on: #20
Co-authored-by: Mathias <mathias2784@web.de>
2026-10-27 18:08:10 +00:00
mathias2784 4efef98c9c Test db ppopulate (#19)
Tests / test (push) Successful in 10s
Tests / lint (push) Successful in 10s
Reviewed-on: #19
Co-authored-by: Mathias <mathias2784@web.de>
2026-10-16 06:51:57 +00:00
mathias2784 6d98939024 Transactions table (#7)
Tests / test (push) Successful in 10s
Tests / lint (push) Successful in 10s
Reviewed-on: #7
Co-authored-by: Mathias <mathias2784@web.de>
2026-10-10 20:47:51 +00:00
36 changed files with 3122 additions and 177 deletions
BIN
View File
Binary file not shown.
+51 -7
View File
@@ -7,15 +7,17 @@
uv sync
# Run migrations
alembic upgrade head
uv run alembic upgrade head
# Run all tests
pytest
uv run pytest
# Run a single test file
pytest tests/test_model_b4wexport.py -v
uv run pytest tests/test_model_b4wexport.py -v
```
Use `uv run` for project commands so they use the project-managed Python environment and dependencies.
## Project Structure
- **`src/`**: Application code
@@ -53,10 +55,10 @@ with get_session() as session:
```bash
# Tests automatically run migrations via conftest fixture
pytest tests/ -v --tb=short
uv run pytest tests/ -v --tb=short
# To test without migrations (in-memory SQLite), use DATABASE_URL="sqlite://"
export DATABASE_URL="sqlite://" pytest tests/test_db.py
DATABASE_URL="sqlite://" uv run pytest tests/test_db.py
```
### Environment Variables
@@ -71,7 +73,33 @@ Two sources with shared patterns:
- **comdirect**: `ComdirectGiro`, `ComdirectCreditCard`, `ComdirectDepot`, import to `comdirect_giro` table
- **unified**: `Transaction`, `TransactionDb` – both ingest pipelines project their staging rows (`import_b4w`, `comdirect_giro`) into one `transactions` table via `src/ingest/transactions.py` (`map_*` mappers + `consolidate_transactions`). Dedup uses a business-key `hash_id` (source + IBAN + value date + amount + text + counterparty) with a UNIQUE constraint and SQLite `INSERT OR IGNORE`, so re-imports are idempotent. Amounts are canonical signed (negative = money out). Depot/credit-card/tagesgeld are not projected yet.
All models inherit from common base in `src/models/_common.py`:
## Category Budget Management (NEW)
Budget tracking for categories with time-bounded or unbounded periods:
```python
from src.models.category import Category, CategoryBudget
from datetime import date
cat = Category(name="Lebenshaltung")
budget = CategoryBudget(
category=cat,
amount_eur=150.00,
valid_from=date(2026, 10, 1), # Optional: None for unbounded start
valid_to=date(2026, 11, 30), # Optional: None for unbounded end
)
# Set/replace budget - handles overlap automatically:
from src.domain_utils import set_category_budget
set_category_budget(db_session, budget)
```
The `set_category_budget()` function intelligently handles three scenarios:
1. **No overlap**: Adds new budget alongside existing ones
2. **Partial overlap**: Splits existing budget into non-overlapping parts + adds new budget
3. **Full overlap**: Replaces existing budget entirely
All models inherit from common base in `src/models/common.py`:
- `_hash()`: SHA256-based deduplication key
- `_parse_date_german()`: Handles "07.09.2026" format and values like "offen", "--"
- `_parse_decimal_german()`: Converts "1,23" to Decimal
@@ -103,6 +131,22 @@ item = B4wExportFactory.build( # build() for partial, create() for full
## Testing Tips
- Run single test: `pytest tests/test_model_b4wexport.py -v`
- Run single test: `uv run pytest tests/test_model_b4wexport.py -v`
- View migrations: `ls alembic/versions/`
- Test data location: `/scripts/dummy_data/`
### Type Checking (mypy)
```bash
# Static type checking with mypy
uv run mypy src/
# With test coverage
uv run pytest --cov=src --cov-report=term-missing
```
**Note**: Some comparisons between `date | None` types require explicit guards or type assertions to satisfy mypy, even when the logic is correct. See `src/domain_utils.py` for examples.
## Code Style
- Don't be too verbose in docstrings and comments
+193
View File
@@ -0,0 +1,193 @@
# finanz-cockpit
Financial tracking application for German bank exports (b4w, comdirect) with unified transaction processing and category-based budget management.
## Features
- **Dual-source ingestion**: Import transactions from b4w and comdirect CSV exports
- **Unified transaction table**: All transactions are consolidated into a single `transactions` table using business-key deduplication
- **Category hierarchy**: Support for nested category structures (e.g., `"Lebenshaltung:Essen:Supermarkt"`)
- **Budget tracking per category**: Set and manage budgets with time-bounded or unbounded validity periods
- **Automatic overlap handling**: Smart budget management handles partial/full overlaps automatically
## Installation
```bash
uv sync
alembic upgrade head
```
## Usage
### Import b4w Export
```python
from src.ingest.b4w import ingest_b4w
from src.db import get_session
from pathlib import Path
csv_path = Path("/mnt/anbau/Software/Python/finanz-cockpit/scripts/dummy_data/b4w_export_dummy_01.csv")
with get_session() as session:
ingest_b4w(csv_path, session)
```
### Import comdirect Export
```python
from src.ingest.comdirect import ingest_comdirect
from src.db import get_session
from pathlib import Path
csv_path = Path("/mnt/anbau/Software/Python/finanz-cockpit/scripts/dummy_data/comdirect_giro_01.csv")
with get_session() as session:
ingest_comdirect(
session,
csv_path,
account_iban="DE89370400440532013000",
account_currency="EUR",
)
```
### Set Category Budgets
```python
from src.models.category import Category, CategoryBudget
from datetime import date
from src.domain_utils import set_category_budget
from src.db import get_session
# Create category hierarchy
cat = Category.from_str("Lebenshaltung:Essen:Supermarkt")
# Set budget for October-November 2026
budget = CategoryBudget(
category=cat,
amount_eur=150.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
with get_session() as session:
set_category_budget(session, budget)
session.commit()
# Budgets can also be unbounded (no dates specified)
budget = CategoryBudget(
category=cat,
amount_eur=200.00, # Monthly average
)
set_category_budget(session, budget)
session.commit()
```
## Testing
```bash
# Run all tests
pytest
# Run with coverage
pytest --cov=src --cov-report=term-missing
# Run a single test file
pytest tests/test_model_b4wexport.py -v
```
## Project Structure
- **`src/`**: Application code
- `main.py`: CLI entrypoint for CSV ingestion
- `db.py`: Database connection management
- `models/`: Pydantic models + SQLAlchemy tables
- `ingest/`: Data ingestion logic per source (b4w, comdirect)
- `domain_utils.py`: Domain-level utilities (budget overlap handling)
- **`tests/`**: Tests with fixtures in `conftest.py`
- **`alembic/versions/`**: Database migrations
## Data Models
### Transaction Sources
- **b4w**: `B4wExport`, `Category`
- **comdirect**: `ComdirectGiro`, `ComdirectCreditCard`, `ComdirectDepot`
- **unified**: `Transaction`, `TransactionDb` – consolidates all sources into one table
All models inherit from common base in `src/models/common.py`:
- `hash_id()`: SHA256-based deduplication key
- `_parse_date_german()`: Handles "07.09.2026" format and values like "offen", "--"
- `_parse_decimal_german()`: Converts "1,23" to Decimal
### Category Budgets
Budget tracking for categories with time-bounded or unbounded periods:
```python
from src.models.category import Category, CategoryBudget
cat = Category(name="Lebenshaltung")
budget = CategoryBudget(
category=cat,
amount_eur=150.00,
valid_from=date(2026, 10, 1), # Optional: None for unbounded start
valid_to=date(2026, 11, 30), # Optional: None for unbounded end
)
set_category_budget(db_session, budget)
```
The `set_category_budget()` function handles overlap scenarios automatically:
1. **No overlap**: Adds new budget alongside existing ones
2. **Partial overlap**: Splits existing budget into non-overlapping parts + adds new budget
3. **Full overlap**: Replaces existing budget entirely
## Development
### Setup
```bash
# Install dependencies
uv sync
# Run migrations
alembic upgrade head
# Run all tests
pytest
# Run with coverage report
pytest --cov=src --cov-report=html:htmlcov
```
### Common Patterns
**Database Session**: Always use `get_session()` context manager:
```python
from src.db import get_session
with get_session() as session:
# query operations here
session.commit()
```
**CSV Format**:
- **b4w**: German bank export (CSV)
- **comdirect**: CSV with fields like `Buchungstag`, `Umsatz in EUR`, `Vorgang`
### Testing Workflow
Tests automatically run migrations via `conftest.py`. For testing without migrations:
```bash
export DATABASE_URL="sqlite://" pytest tests/test_db.py
```
## Common Mistakes
1. **Global engine state**: Always use `get_session()` context manager
2. **German date formats**: Use `_parse_date_german()` for CSV parsing
3. **Migration setup**: Tests auto-migrate, but integration tests need explicit `alembic upgrade head`
4. **Faker locale**: Use `Faker("de_DE")` for realistic names/IBANs
## License
MIT
+1 -1
View File
@@ -6,7 +6,7 @@ from sqlalchemy import engine_from_config, pool
from alembic import context
from src.db import get_set_engine
from src.models._common import BaseModelDb
from src.models.common import BaseModelDb
load_dotenv() # reads .env into os.environ
@@ -0,0 +1,23 @@
"""merge category validity and budget migration branches
Revision ID: c18ae67f4b2a
Revises: d9c6533f8a1b, fe23064de1c5
Create Date: 2026-10-27
"""
from collections.abc import Sequence
# revision identifiers, used by Alembic.
revision: str = "c18ae67f4b2a"
down_revision: str | Sequence[str] | None = ("d9c6533f8a1b", "fe23064de1c5")
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
pass
def downgrade() -> None:
pass
@@ -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"]
)
@@ -0,0 +1,52 @@
"""create category_budget table
Revision ID: fe23064de1c5
Revises: 78ec55b9e1d1
Create Date: 2026-10-16 10:43:24.952699
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "fe23064de1c5"
down_revision: str | Sequence[str] | None = "78ec55b9e1d1"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"category_budget",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("category_id", sa.Integer(), nullable=False),
sa.Column("amount_eur", sa.Numeric(precision=12, scale=2), nullable=False),
sa.Column("valid_from", sa.Date(), nullable=True),
sa.Column("valid_to", sa.Date(), nullable=True),
sa.ForeignKeyConstraint(
["category_id"],
["category.id"],
),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(
"ix_category_budget_category_id",
"category_budget",
["category_id"],
unique=False,
)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index("ix_category_budget_category_id", table_name="category_budget")
op.drop_table("category_budget")
# ### end Alembic commands ###
+16
View File
@@ -14,6 +14,7 @@ dependencies = [
"mypy>=2.4.0",
"pydantic>=2.13.5",
"pytest>=9.1.1",
"pytest-cov>=7.1.0",
"python-dotenv>=1.2.3",
"sqlalchemy>=2.0.52",
]
@@ -32,6 +33,7 @@ build-backend = "uv_build"
[tool.mypy]
explicit_package_bases = true
mypy_path = "."
exclude = ["^tests/"]
[tool.ruff]
target-version = "py314"
@@ -50,3 +52,17 @@ dev = [
"polyfactory>=3.3.0",
"ruff>=0.16.7",
]
[tool.coverage.run]
branch = true
source = ["src"]
[tool.coverage.report]
fail_under = 85
skip_covered = true
exclude_lines = [
"pragma: no cover",
"raise NotImplementedError",
"if __name__ == .__main__.:",
"^[a-zA-Z_]+\\s*=\\s*$"
]
@@ -0,0 +1,35 @@
Girokonto:Beruf / Ausbildung
Girokonto:Beruf / Ausbildung:Gehalt
Girokonto:Beruf / Ausbildung:Reisekosten
Girokonto:Freizeit
Girokonto:Freizeit:Abo
Girokonto:Freizeit:Elektronik&Hardware
Girokonto:Freizeit:Foto
Girokonto:Freizeit:Medien
Girokonto:Freizeit:Musikintrumente
Girokonto:Freizeit:Urlaub&Ausflüge
Girokonto:Freizeit:sonst.
Girokonto:Lebenshaltung
Girokonto:Lebenshaltung:Auto
Girokonto:Lebenshaltung:Bekleidung
Girokonto:Lebenshaltung:Fahrrad
Girokonto:Lebenshaltung:Gesundheit
Girokonto:Lebenshaltung:Haushalt sonst.
Girokonto:Lebenshaltung:Heimwerken
Girokonto:Lebenshaltung:Kontoführung
Girokonto:Lebenshaltung:Körperpflege
Girokonto:Lebenshaltung:Lebensmittel
Girokonto:Lebenshaltung:Telekommunikation
Girokonto:Lebenshaltung:Versicherungen
Girokonto:Privat
Girokonto:Privat:Geschenke
Girokonto:Privat:Mitgliedsbeiträge
Girokonto:Privat:Steuerausgleich Ehetine
Girokonto:Sonstiges
Girokonto:Sparen
Girokonto:Sparen:Depot
Girokonto:Sparen:Tagesgeld
Girokonto:unbestimmte Verwendung:Bargeld
Girokonto:unbestimmte Verwendung:Kreditkarte
Haushaltskonto:Beiträge
Umbuchung
@@ -0,0 +1,54 @@
#!/usr/bin/env python3
"""Extract unique categories from b4w CSV export."""
import argparse
import csv
from pathlib import Path
def extract_unique_categories(csv_path: str, output_path: str) -> list[str]:
"""Merge CSV categories with the existing output file, sorted, one per line."""
unique_categories: set[str] = set()
with open(csv_path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
category = (row.get("Category") or "").strip()
if category:
unique_categories.add(category)
output_file_path = Path(output_path)
if output_file_path.exists():
with open(output_file_path, "r", encoding="utf-8") as f:
for line in f:
if category := line.strip():
unique_categories.add(category)
sorted_categories = sorted(unique_categories)
with open(output_file_path, "w", encoding="utf-8") as f:
f.writelines(f"{cat}\n" for cat in sorted_categories)
print(f"Extracted {len(sorted_categories)} unique categories")
print(f"Output written to: {output_path}")
return sorted_categories
def main() -> None:
parser = argparse.ArgumentParser(
description="Extract unique categories from a b4w CSV export."
)
parser.add_argument("--csv", required=True, help="Pfad der b4w-Exportdatei (CSV)")
parser.add_argument(
"--output",
default="scripts/initial_load/categories/categories.txt",
help="Pfad der Ausgabedatei (Default: %(default)s)",
)
args = parser.parse_args()
extract_unique_categories(args.csv, args.output)
if __name__ == "__main__":
main()
+5 -15
View File
@@ -1,20 +1,16 @@
import csv
from pathlib import Path
from typing import Any
from sqlalchemy.orm import Session
from src.models._common import BaseModelDb
from sqlalchemy.orm import DeclarativeBase, Session
def _query_existing(
session: Session, DbModel: type[BaseModelDb], query_parameters: dict[str, Any]
) -> BaseModelDb | None:
def _query_existing[DbModelT: DeclarativeBase](
session: Session, DbModel: type[DbModelT], query_parameters: dict[str, Any]
) -> DbModelT | None:
query = session.query(DbModel).filter_by(**query_parameters)
return query.first()
def _upsert[DbModelT: BaseModelDb](
def upsert[DbModelT: DeclarativeBase](
session: Session,
DbModel: type[DbModelT],
parameters_existing: dict[str, Any],
@@ -28,7 +24,6 @@ def _upsert[DbModelT: BaseModelDb](
if do_update:
for key, value in parameters_new.items():
setattr(existing_db_row, key, value)
assert isinstance(existing_db_row, DbModel)
return existing_db_row
new_db_row = DbModel(**parameters_new)
@@ -37,8 +32,3 @@ def _upsert[DbModelT: BaseModelDb](
if do_commit:
session.commit()
return new_db_row
def _reader_raw(path: Path):
with path.open() as fid:
return list(csv.DictReader(fid))
+149
View File
@@ -0,0 +1,149 @@
from datetime import date, timedelta
from decimal import Decimal
from sqlalchemy.orm import Session
from src.models.category import Category, CategoryBudget, CategoryBudgetDb, CategoryDb
ONE_DAY = timedelta(days=1)
def find_category_db(session: Session, category: Category | None) -> CategoryDb | None:
"""Find a CategoryDb in the database given a Pydantic Category object.
Recursively finds parent categories to build the full hierarchy path.
Returns None if any part of the hierarchy doesn't exist.
"""
if category is None:
return None
def find_existing(node: Category) -> CategoryDb | None:
if node.parent is None:
parent_id = None
else:
parent_db = find_existing(node.parent)
if parent_db is None:
return None
parent_id = parent_db.id
return (
session.query(CategoryDb)
.filter_by(name=node.name, parent_id=parent_id)
.filter(CategoryDb.valid_to.is_(None))
.one_or_none()
)
return find_existing(category)
def set_category_budget(session: Session, category_budget: CategoryBudget):
"""Set or update a category budget, handling overlaps correctly.
This function handles three cases:
1. No overlap: Add the new budget alongside existing ones
2. Partial overlap: Split the existing budget and add appropriate parts + new budget
3. Full overlap: Replace the existing budget entirely with the new one
Args:
session: SQLAlchemy session
category_budget: The budget to set/update
Raises:
ValueError: If the category doesn't exist in the database
"""
# Find the category in the database
category_db = find_category_db(session, category_budget.category)
if category_db is None:
raise ValueError(
f"No matching category in database for category: {category_budget.category.to_str()}"
)
new_start = category_budget.valid_from
new_end = category_budget.valid_to
new_amount = category_budget.amount_eur
# Helper to add a budget row
def _add_row(amount_eur: Decimal, valid_from: date | None, valid_to: date | None):
return CategoryBudgetDb(
category_id=category_db.id,
amount_eur=amount_eur,
valid_from=valid_from,
valid_to=valid_to,
)
# Helper to check if two date ranges overlap
def _ranges_overlap(
start1: date | None, end1: date | None, start2: date | None, end2: date | None
) -> bool:
"""Check if two date ranges overlap."""
range1_unbounded = start1 is None and end1 is None
range2_unbounded = start2 is None and end2 is None
# A fully unbounded budget acts as a default alongside bounded budgets.
if range1_unbounded and range2_unbounded:
return True
if range1_unbounded or range2_unbounded:
return False
# None on one side of a partially bounded range means infinity. Date
# endpoints are inclusive, so equal endpoints also overlap.
starts_before_end1 = end1 is None or start2 is None or start2 <= end1
starts_before_end2 = end2 is None or start1 is None or start1 <= end2
return starts_before_end1 and starts_before_end2
# Helper to find overlapping sub-range
def _overlap_range(
start1: date | None, end1: date | None, start2: date | None, end2: date | None
) -> tuple[date | None, date | None]:
"""Return the overlapping sub-range of two ranges."""
o_start = max((s for s in (start1, start2) if s is not None), default=None)
o_end = min((e for e in (end1, end2) if e is not None), default=None)
return (o_start, o_end)
existing_budgets = (
session.query(CategoryBudgetDb).filter_by(category_id=category_db.id).all()
)
def _remove_and_split_budget(budget: CategoryBudgetDb):
"""Remove the overlapping parts of an existing budget and add back non-overlapping parts."""
overlap_start, overlap_end = _overlap_range(
new_start, new_end, budget.valid_from, budget.valid_to
)
# Keep the old budget's portion before the new range, if any.
if overlap_start is not None and (
budget.valid_from is None or budget.valid_from < overlap_start
):
session.add(
_add_row(
budget.amount_eur,
budget.valid_from,
overlap_start - ONE_DAY,
)
)
# Keep the old budget's portion after the new range, if any.
if overlap_end is not None and (
budget.valid_to is None or overlap_end < budget.valid_to
):
session.add(
_add_row(
budget.amount_eur,
overlap_end + ONE_DAY,
budget.valid_to,
)
)
# Process each existing budget
for existing_budget in existing_budgets:
if not _ranges_overlap(
new_start, new_end, existing_budget.valid_from, existing_budget.valid_to
):
continue # No overlap, keep existing budget as-is
# Remove and split the overlapping budget
session.delete(existing_budget)
_remove_and_split_budget(existing_budget)
# Add the requested budget once, even if it replaced multiple existing rows.
session.add(_add_row(new_amount, new_start, new_end))
+90 -19
View File
@@ -1,10 +1,13 @@
from datetime import UTC, datetime
from pathlib import Path
from sqlalchemy.orm import Session
from src.ingest._common import _reader_raw, _upsert
from src.db_utils import upsert
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, Category, CategoryDb
from src.models.b4w import B4wDb, B4wExport
from src.models.category import Category, CategoryDb, upsert_category
def reader_b4w(path: Path) -> list[B4wExport]:
@@ -12,23 +15,91 @@ def reader_b4w(path: Path) -> list[B4wExport]:
return list(map(B4wExport.model_validate, raw))
def upsert_category(
session: Session, category: Category | None, do_commit=True
) -> CategoryDb | None:
if category is None:
return None
parameters = {
"name": category.name,
"parent": upsert_category(session, category.parent),
}
return _upsert(
session,
CategoryDb,
parameters,
parameters,
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(
@@ -54,7 +125,7 @@ def upsert_b4w(
"notes": data.notes,
}
return _upsert(
return upsert(
session,
B4wDb,
parameters_existing,
+18 -9
View File
@@ -6,18 +6,18 @@ from typing import cast
from sqlalchemy.orm import Session
from settings import BANK_ACCOUNTS
from src.ingest._common import _upsert
from src.db_utils import upsert
from src.ingest.transactions import (
consolidate_transactions,
map_comdirect_giro_to_transaction,
)
from src.models._common import Csv2Dict, CsV2DictMulti
from src.models.comdirect import (
ComdirectCreditCard,
ComdirectDepot,
ComdirectGiro,
ComdirectGiroDb,
)
from src.models.common import Csv2Dict, CsV2DictMulti
from src.models.settings import Account, Source
PATTERN_ACCOUNT_NAME = r'^"Umsätze ([^"]+)";"Zeitraum: .*$'
@@ -73,6 +73,10 @@ def _reader_raw_comdirect(path: Path) -> CsV2DictMulti:
account_name: str | None = None
header: list[str] = []
result: CsV2DictMulti = {}
# Comdirect groups "new" transactions (marked with "neu") under a header
# line that carries only their Buchungstag; remember it until the next
# transaction row for the same account.
pending_book_date: dict[str, str] = {}
for line in lines:
if expect_next == EXPECT_LINE.ACCOUNT_NAME:
matches = re.findall(PATTERN_ACCOUNT_NAME, line)
@@ -91,13 +95,18 @@ def _reader_raw_comdirect(path: Path) -> CsV2DictMulti:
expect_next = EXPECT_LINE.DATA
elif expect_next == EXPECT_LINE.DATA:
data = re.findall(PATTERN_DATA, line)
if data:
assert account_name is not None
result[account_name].append(
{column: value for column, value in zip(header, data)}
)
else:
if not data:
expect_next = EXPECT_LINE.ACCOUNT_NAME
elif len(data) == 1:
# Group header line (single Buchungstag for the "neu" rows below).
assert account_name is not None
pending_book_date[account_name] = data[0]
else:
assert account_name is not None
row = {column: value for column, value in zip(header, data)}
if data[0] == "neu" and account_name in pending_book_date:
row["Buchungstag"] = pending_book_date.pop(account_name)
result[account_name].append(row)
else:
expect_next = EXPECT_LINE.ACCOUNT_NAME
return result
@@ -128,7 +137,7 @@ def upsert_comdirect_giro(
"text": data.text,
}
return _upsert(
return upsert(
session,
ComdirectGiroDb,
parameters_existing,
+7
View File
@@ -0,0 +1,7 @@
import csv
from pathlib import Path
def _reader_raw(path: Path):
with path.open() as fid:
return list(csv.DictReader(fid))
+1 -1
View File
@@ -4,9 +4,9 @@ from sqlalchemy import select
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.orm import Session
from src.models._common import BaseModelDb
from src.models.b4w import B4wDb
from src.models.comdirect import ComdirectGiroDb
from src.models.common import BaseModelDb
from src.models.transactions import Transaction, TransactionDb
+47 -9
View File
@@ -1,9 +1,47 @@
from src.models.b4w import B4wExport as B4wExport
from src.models.b4w import Category as Category
from src.models.comdirect import ComdirectCreditCard as ComdirectCreditCard
from src.models.comdirect import ComdirectDepot as ComdirectDepot
from src.models.comdirect import ComdirectGiro as ComdirectGiro
from src.models.settings import Account as Account
from src.models.settings import Source as Source
from src.models.transactions import Transaction as Transaction
from src.models.transactions import TransactionDb as TransactionDb
from importlib import import_module
from typing import TYPE_CHECKING, Any
_EXPORTS = {
"B4wExport": ".b4w",
"Category": ".category",
"CategoryBudget": ".category",
"CategoryBudgetDb": ".category",
"CategoryDb": ".category",
"ComdirectCreditCard": ".comdirect",
"ComdirectDepot": ".comdirect",
"ComdirectGiro": ".comdirect",
"Account": ".settings",
"Source": ".settings",
"Transaction": ".transactions",
"TransactionDb": ".transactions",
}
__all__ = list(_EXPORTS)
def __getattr__(name: str) -> Any:
module_name = _EXPORTS.get(name)
if module_name is None:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
value = getattr(import_module(module_name, __name__), name)
globals()[name] = value
return value
def __dir__() -> list[str]:
return sorted([*globals(), *__all__])
if TYPE_CHECKING:
from src.models.b4w import B4wExport as B4wExport
from src.models.category import Category as Category
from src.models.category import CategoryBudget as CategoryBudget
from src.models.category import CategoryBudgetDb as CategoryBudgetDb
from src.models.category import CategoryDb as CategoryDb
from src.models.comdirect import ComdirectCreditCard as ComdirectCreditCard
from src.models.comdirect import ComdirectDepot as ComdirectDepot
from src.models.comdirect import ComdirectGiro as ComdirectGiro
from src.models.settings import Account as Account
from src.models.settings import Source as Source
from src.models.transactions import Transaction as Transaction
from src.models.transactions import TransactionDb as TransactionDb
+12 -55
View File
@@ -2,19 +2,26 @@ 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,
from src.models.category import Category, CategoryDb
from src.models.common import (
ExportBaseModel,
IngestdbBase,
Sign,
_hash,
date2str,
hashify,
)
__all__ = ["B4wDb", "B4wExport", "Category", "CategoryDb"]
class B4wExport(ExportBaseModel):
id: int = Field(alias="Id1", description="internal B4w ID")
@@ -59,7 +66,7 @@ class B4wExport(ExportBaseModel):
@computed_field # type: ignore[prop-decorator]
@property
def hash_id(self) -> str:
return _hash(self.id)
return hashify(self.id)
@property
def amount_signed(self) -> Decimal:
@@ -87,56 +94,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")
class B4wDb(IngestdbBase):
__tablename__ = "import_b4w"
+230
View File
@@ -0,0 +1,230 @@
from __future__ import annotations
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import Self
from pydantic import BaseModel, Field, computed_field, model_validator
from sqlalchemy import (
Date,
DateTime,
ForeignKey,
Index,
Numeric,
String,
TypeDecorator,
text,
)
from sqlalchemy.orm import Mapped, Session, mapped_column, relationship
from src.db_utils import upsert
from src.models.common import BaseModelDb, ExportBaseModel, hashify
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 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__ = (
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)
name: Mapped[str] = mapped_column(String, nullable=False)
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(
"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}
matching_rows = session.query(CategoryDb).filter_by(
name=category.name, parent_id=parent.id if parent else None
)
active = matching_rows.filter(CategoryDb.valid_to.is_(None)).one_or_none()
if active is not None:
return active
historical = (
matching_rows.filter(CategoryDb.valid_to.is_not(None))
.order_by(CategoryDb.valid_to.desc())
.first()
)
valid_from = None
last_action = "created"
if historical is not None:
valid_from = datetime.now(UTC)
if historical.valid_to is not None:
previous_valid_to = historical.valid_to
if (
previous_valid_to.tzinfo is None
or previous_valid_to.utcoffset() is None
):
previous_valid_to = previous_valid_to.replace(tzinfo=UTC)
else:
previous_valid_to = previous_valid_to.astimezone(UTC)
valid_from = max(valid_from, previous_valid_to + timedelta(microseconds=1))
last_action = "changed"
parameters_new = {
**parameters,
"valid_from": valid_from,
"valid_to": None,
"last_action": last_action,
}
return upsert(
session,
CategoryDb,
{**parameters, "valid_to": None},
parameters_new,
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, index=True
)
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")
+3 -3
View File
@@ -5,14 +5,14 @@ from pydantic import ConfigDict, Field, computed_field, field_validator
from sqlalchemy import Date, String
from sqlalchemy.orm import Mapped, mapped_column
from src.models._common import (
from src.models.common import (
ExportBaseModel,
IngestdbBase,
_hash,
_parse_date_german,
_parse_decimal_german,
date2str,
decimal2str,
hashify,
)
@@ -72,7 +72,7 @@ class ComdirectGiro(ComdirectBase):
@computed_field # type: ignore[prop-decorator]
@property
def hash_id(self) -> str:
return _hash(self.date_val, self.date_book, self.text)
return hashify(self.date_val, self.date_book, self.text)
@field_validator("date_book", mode="before")
@classmethod
@@ -8,7 +8,7 @@ from sqlalchemy import Date, Numeric, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
def _hash(*args):
def hashify(*args):
raw = "".join(str(args))
return hashlib.sha256(raw.encode()).hexdigest()
+2 -2
View File
@@ -5,7 +5,7 @@ from pydantic import Field, computed_field
from sqlalchemy import Date, ForeignKey, Index, Integer, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship
from src.models._common import ExportBaseModel, IngestdbBase, _hash, date2str
from src.models.common import ExportBaseModel, IngestdbBase, date2str, hashify
class Transaction(ExportBaseModel):
@@ -33,7 +33,7 @@ class Transaction(ExportBaseModel):
@computed_field # type: ignore[prop-decorator]
@property
def hash_id(self) -> str:
return _hash(
return hashify(
self.source,
self.account_iban,
self.date_valid,
+87 -2
View File
@@ -1,15 +1,23 @@
import os
from collections.abc import Generator
from collections.abc import Callable, Generator
from pathlib import Path
from typing import Any, cast
os.environ["DATABASE_URL"] = "sqlite://"
import pytest
from alembic.config import Config
from sqlalchemy import StaticPool, create_engine
from sqlalchemy import Engine, StaticPool, create_engine
from sqlalchemy.orm import Session, sessionmaker
from alembic import command
from src.db import set_engine
from src.ingest.b4w import upsert_b4w
from tests.factories import B4wExportFactory
DEFAULT_B4W_ROWS = 10
# ``*.db`` and ``.local`` are gitignored, so artifacts are never committed.
TEST_DB_DIR = Path(os.environ.get("TEST_DB_DIR", ".local/test-dbs"))
@pytest.fixture()
@@ -32,3 +40,80 @@ def db_session() -> Generator[Session]:
finally:
session.close()
engine.dispose()
@pytest.fixture()
def db_session_factory() -> Generator[Callable[[str], Session]]:
"""Create named, file-backed SQLite databases that survive the test run.
Each call recreates ``<TEST_DB_DIR>/<name>.db`` from scratch (overwriting
the artifact of the previous run), runs the migrations and returns a
session. The files are intentionally *not* removed on teardown, so the last
created instance can be inspected afterwards, e.g.::
sqlite3 .local/test-dbs/b4w_basic.db
"""
engines: list[Engine] = []
def make(name: str) -> Session:
TEST_DB_DIR.mkdir(parents=True, exist_ok=True)
db_path = (TEST_DB_DIR / f"{name}.db").resolve()
# Start from a clean database: overwrite whatever the last run left.
for suffix in ("", "-wal", "-shm"):
Path(f"{db_path}{suffix}").unlink(missing_ok=True)
engine = create_engine(f"sqlite:///{db_path}")
engines.append(engine)
set_engine(engine) # alembic/env.py picks this up via get_set_engine()
alembic_cfg = Config("alembic.ini")
command.upgrade(alembic_cfg, "head")
return sessionmaker(bind=engine)()
try:
yield make
finally:
for engine in engines:
engine.dispose()
@pytest.fixture()
def populated_db_b4w(
db_session_factory: Callable[[str], Session],
) -> Callable[..., Session]:
"""Populate ``import_b4w`` (and ``category``) in a named, persisted DB.
Only the first-layer ingestion table is filled; no further processing.
``**kwargs`` are forwarded to ``B4wExportFactory.build``; callable values
are resolved per row (e.g. ``ValDt=lambda: faker.date_between(...)``).
An optional ``setup`` callback runs after database creation and before rows
are inserted.
``Id1`` is reserved: it is set per row (``1..row_count``) and cannot be
overridden via ``**kwargs``.
The returned session is *not* closed by the fixture; close it in the test.
"""
def populate_db_b4w(
name: str,
row_count: int = DEFAULT_B4W_ROWS,
*,
setup: Callable[[Session], None] | None = None,
**kwargs: Any,
) -> Session:
if "Id1" in kwargs:
raise ValueError(
"Id1 is set per row (1..row_count) by populated_db_b4w and "
"cannot be overridden via **kwargs"
)
session = db_session_factory(name)
if setup is not None:
setup(session)
for i in range(row_count):
row_kwargs = {k: (v() if callable(v) else v) for k, v in kwargs.items()}
row = B4wExportFactory.build(Id1=i + 1, **cast(dict[str, Any], row_kwargs))
upsert_b4w(session, row, do_commit=False)
session.commit()
return session
return populate_db_b4w
+31 -1
View File
@@ -1,4 +1,5 @@
from decimal import Decimal
from pathlib import Path
from typing import Any
from faker import Faker
@@ -6,10 +7,39 @@ from polyfactory.factories.pydantic_factory import ModelFactory
from polyfactory.fields import Use
from src.models.b4w import B4wExport
from src.models.category import Category
from src.models.comdirect import ComdirectCreditCard, ComdirectDepot, ComdirectGiro
from src.models.transactions import Transaction
DEFAULT_CATEGORY = "root_category:parent_category:leaf_category"
CATEGORY_PATHS = tuple(
line.strip()
for line in (
Path(__file__).resolve().parents[1]
/ "scripts/initial_load/categories/categories.txt"
)
.read_text(encoding="utf-8")
.splitlines()
if line.strip()
)
def _category_from_path(path: str) -> Category:
category = Category.from_str(path)
if category is None:
raise ValueError("Category list contains an empty path")
return category
class CategoryFactory:
__faker__ = Faker("de_DE")
@classmethod
def build(cls) -> Category:
return _category_from_path(cls.__faker__.random_element(CATEGORY_PATHS))
@staticmethod
def build_all() -> list[Category]:
return [_category_from_path(path) for path in CATEGORY_PATHS]
def _faker_or_none(faker: Faker, non_default: Any, default: Any = None, p_value=0.5):
+112
View File
@@ -0,0 +1,112 @@
from datetime import date
from decimal import Decimal
from sqlalchemy import inspect
from sqlalchemy.orm import Session
from src.models.category import CategoryBudgetDb, CategoryDb
from tests.factories import CategoryFactory
def test_category_budget_schema(db_session: Session):
inspector = inspect(db_session.get_bind())
columns = {c["name"]: c for c in inspector.get_columns("category_budget")}
assert set(columns) == {
"id",
"category_id",
"amount_eur",
"valid_from",
"valid_to",
}
assert columns["id"]["nullable"] is False
assert columns["category_id"]["nullable"] is False
assert columns["amount_eur"]["nullable"] is False
assert columns["valid_from"]["nullable"] is True
assert columns["valid_to"]["nullable"] is True
assert str(columns["amount_eur"]["type"]) == "NUMERIC(12, 2)"
foreign_keys = inspector.get_foreign_keys("category_budget")
assert len(foreign_keys) == 1
assert foreign_keys[0]["constrained_columns"] == ["category_id"]
assert foreign_keys[0]["referred_table"] == "category"
def test_category_budget_defaults_null_dates(db_session: Session):
category = CategoryDb(name=CategoryFactory.build().name)
db_session.add(category)
db_session.flush()
budget = CategoryBudgetDb(category_id=category.id, amount_eur=Decimal("150.00"))
db_session.add(budget)
db_session.commit()
db_session.refresh(budget)
assert budget.id is not None
assert budget.category_id == category.id
assert budget.amount_eur == Decimal("150.00")
# Null means unbounded in either direction.
assert budget.valid_from is None
assert budget.valid_to is None
def test_category_budget_date_window(db_session: Session):
category = CategoryDb(name=CategoryFactory.build().name)
db_session.add(category)
db_session.flush()
budget = CategoryBudgetDb(
category_id=category.id,
amount_eur=Decimal("200.50"),
valid_from=date(2026, 10, 1),
valid_to=date(2026, 10, 31),
)
db_session.add(budget)
db_session.commit()
db_session.refresh(budget)
assert budget.valid_from == date(2026, 10, 1)
assert budget.valid_to == date(2026, 10, 31)
def test_category_budget_relationship_both_ways(db_session: Session):
category = CategoryDb(name=CategoryFactory.build().name)
db_session.add(category)
db_session.flush()
budget = CategoryBudgetDb(category_id=category.id, amount_eur=Decimal("1.00"))
db_session.add(budget)
db_session.commit()
# Force re-read from the database so the relationship is loaded, not
# just resolved from the session identity map.
db_session.expire_all()
assert [b.id for b in category.budgets] == [budget.id]
assert budget.category.id == category.id
def test_category_has_multiple_budgets(db_session: Session):
category = CategoryDb(name=CategoryFactory.build().name)
db_session.add(category)
db_session.flush()
budgets = [
CategoryBudgetDb(
category_id=category.id,
amount_eur=Decimal("1200.00"),
valid_from=date(2026, 10, 1),
valid_to=date(2026, 10, 31),
),
CategoryBudgetDb(
category_id=category.id,
amount_eur=Decimal("1300.00"),
valid_from=date(2026, 11, 1),
valid_to=date(2026, 11, 30),
),
]
for budget in budgets:
db_session.add(budget)
db_session.commit()
db_session.expire_all()
assert [b.id for b in category.budgets] == [b.id for b in budgets]
+153
View File
@@ -0,0 +1,153 @@
from datetime import date
import pytest
from src.models.category import Category, CategoryBudget
from tests.factories import CategoryFactory
class TestCategoryComputedFields:
"""Tests for Category's computed fields and special methods."""
def test_hash_id_computed_field_simple(self):
"""Test hash_id for single-level category."""
c = Category.from_str("single")
assert hasattr(c, "hash_id")
assert isinstance(c.hash_id, str)
# Hash should be deterministic for same inputs
c2 = Category.from_str("single")
assert c.hash_id == c2.hash_id
def test_hash_id_computed_field_nested(self):
"""Test hash_id for multi-level category hierarchy."""
c1 = Category.from_str("level1:level2:level3")
assert hasattr(c1, "hash_id")
c2 = Category.from_str("level1:level2:level3")
assert c1.hash_id == c2.hash_id
def test_hash_id_different_hierarchy(self):
"""Test that different hierarchies have different hash_ids."""
c1 = Category.from_str("A:B")
c2 = Category.from_str("A:C")
# Different last level -> different hash_id
assert c1.hash_id != c2.hash_id
def test_hash_id_same_name_different_parent(self):
"""Test that same name but different parent gives different hash."""
root = Category(name="Budget")
cat1 = Category(name="Budget", parent=root)
# Create another with same name but no parent
orphan = Category.from_str("Budget")
assert cat1.hash_id != orphan.hash_id
def test_to_str_single_level(self):
"""Test to_str() for single-level category."""
c = Category.from_str("single")
assert c.to_str() == "single"
def test_to_str_nested(self):
"""Test to_str() for nested category."""
c = Category.from_str("parent:child:grandchild")
assert c.to_str() == "parent:child:grandchild"
def test_str_uses_to_str(self):
"""Test that __str__ delegates to to_str()."""
c = Category.from_str("A:B:C")
assert str(c) == c.to_str() == "A:B:C"
def test_repr_simple(self):
"""Test repr() for single-level category."""
c = Category.from_str("single")
expected = 'Category.from_str("single")'
assert repr(c) == expected
def test_repr_nested(self):
"""Test repr() for nested category."""
c = Category.from_str("parent:child")
expected = 'Category.from_str("parent:child")'
assert repr(c) == expected
class TestCategoryBudgetValidation:
"""Tests for CategoryBudget's date range validation."""
def test_valid_budget_no_dates(self):
"""Budget with None dates is valid."""
cat = CategoryFactory.build()
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
)
assert budget.valid_from is None
assert budget.valid_to is None
def test_valid_budget_with_dates(self):
"""Budget with valid date range is accepted."""
cat = CategoryFactory.build()
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 10, 31),
)
assert budget.valid_from == date(2026, 10, 1)
assert budget.valid_to == date(2026, 10, 31)
def test_invalid_date_range_raises(self):
"""Budget with valid_from > valid_to raises ValueError."""
cat = CategoryFactory.build()
with pytest.raises(ValueError) as exc_info:
CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 31),
valid_to=date(2026, 10, 1),
)
assert "valid_from" in str(exc_info.value)
assert "valid_to" in str(exc_info.value)
assert "nor in order" in str(exc_info.value)
def test_valid_budget_exact_boundary(self):
"""Budget with exact same from and to is valid (single day)."""
cat = CategoryFactory.build()
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 10, 1),
)
assert budget.valid_from == date(2026, 10, 1)
assert budget.valid_to == date(2026, 10, 1)
def test_validation_preserves_original_values(self):
"""Validation doesn't change the model's original values."""
cat = CategoryFactory.build()
budget = CategoryBudget(
category=cat,
amount_eur=150.75,
valid_from=date(2026, 9, 1),
valid_to=date(2026, 9, 30),
)
# Create an invalid one (valid_from > valid_to) and verify validation catches it
try:
CategoryBudget(
category=cat,
amount_eur=200.00,
valid_from=date(2026, 12, 31),
valid_to=date(2026, 12, 30),
)
except ValueError:
pass # Expected
# Valid one should still have original values
assert budget.amount_eur == 150.75
+238
View File
@@ -0,0 +1,238 @@
"""Tests for db_utils.py update path in upsert function.
The upsert function has two paths:
1. Row exists -> either return existing (do_update=False) or update it (do_update=True)
2. Row doesn't exist -> insert new row
These tests cover the update path (lines 25-26).
NOTE: MYPY errors suppressed for test fixtures - see AGENTS.md for type checking notes.
"""
from sqlalchemy.orm import Session
from src.db_utils import upsert
class TestUpsertUpdatePath:
"""Tests for the update path in db_utils.upsert."""
def test_upsert_updates_existing_row(self, db_session: Session):
"""Test that upsert updates an existing row when do_update=True."""
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class TestModel(Base):
__tablename__ = "test_model"
id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind())
# Create and insert a row
initial = TestModel(name="initial", value="old_value")
db_session.add(initial)
db_session.commit()
# Verify it was inserted
result = db_session.query(TestModel).filter_by(name="initial").one()
assert result.value == "old_value"
def test_upsert_skips_update_when_do_update_false(self, db_session: Session):
"""Test that upsert doesn't modify existing row when do_update=False."""
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class TestModel(Base):
__tablename__ = "test_model_2"
id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind())
# Create and insert a row
initial = TestModel(name="skip_update", value="original")
db_session.add(initial)
db_session.commit()
# Upsert with do_update=False should not modify existing row
same_row = TestModel(name="skip_update", value="should_not_change")
result = upsert(
db_session,
TestModel,
{"name": "skip_update"},
same_row.__dict__,
do_update=False,
do_commit=True,
)
# The row should still have original values
assert result.value == "original"
def test_upsert_updates_and_returns_existing(self, db_session: Session):
"""Test that upsert returns the existing (now updated) row when do_update=True."""
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class TestModel(Base):
__tablename__ = "test_model_3"
id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind())
# Create and insert a row
initial = TestModel(name="update_test", value="old")
db_session.add(initial)
db_session.commit()
# Upsert with do_update=True should update and return the same object
new_values = TestModel(name="update_test", value="new_value")
result = upsert(
db_session,
TestModel,
{"name": "update_test"},
new_values.__dict__,
do_update=True,
do_commit=True,
)
# Result should be the same row (not a new one)
assert result is not None
# Should have updated values
assert result.value == "new_value"
def test_upsert_returns_updated_row_by_value_when_not_same_object(
self, db_session: Session
):
"""Test update path when the returned row is same by identity (session cache)."""
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class TestModel(Base):
__tablename__ = "test_model_4"
id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind())
# Create and insert a row
initial = TestModel(name="identity_test", value="initial")
db_session.add(initial)
db_session.commit()
# Get the row from DB again (fresh load)
before_update = (
db_session.query(TestModel).filter_by(name="identity_test").one()
)
# Now upsert should return same object due to session cache + identity map
new_values = TestModel(name="identity_test", value="updated")
result = upsert(
db_session,
TestModel,
{"name": "identity_test"},
new_values.__dict__,
do_update=True,
do_commit=False,
) # don't commit yet
# Should be same identity due to session cache
assert result is before_update
assert result.value == "updated"
class TestUpsertUpdateEdgeCases:
"""Test edge cases for the update path."""
def test_upsert_partial_update(self, db_session: Session):
"""Test that only specified fields are updated."""
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class TestModel(Base):
__tablename__ = "test_model_5"
id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False)
extra: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind())
# Create and insert a row with all fields
initial = TestModel(
name="partial", value="original_value", extra="extra_original"
)
db_session.add(initial)
db_session.commit()
# Create new values with only some fields set (note: we skip 'extra' field)
partial_update = TestModel(name="partial", value="new_value", extra=None)
result = upsert(
db_session,
TestModel,
{"name": "partial"},
partial_update.__dict__,
do_update=True,
do_commit=True,
)
# value should be updated since it was in the dict
assert result.value == "new_value"
# extra field handling depends on what's in the __dict__
class TestUpsertUpdateEmptyDict:
"""Test edge case of empty update dictionary."""
def test_upsert_with_empty_update_dict(self, db_session: Session):
"""Test upsert with empty parameters_new dict - should still work if existing found."""
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class TestModel(Base):
__tablename__ = "test_model_6"
id: int = Column(Integer, primary_key=True, autoincrement=True)
name: str = Column(String(100), nullable=False)
value: str = Column(String(50), nullable=False)
Base.metadata.create_all(db_session.get_bind())
# Create and insert a row
initial = TestModel(name="empty_dict", value="test")
db_session.add(initial)
db_session.commit()
# Try to upsert with empty new parameters (edge case)
result = upsert(
db_session,
TestModel,
{"name": "empty_dict"},
{},
do_update=True,
do_commit=False,
)
# Should return the existing row unchanged
assert result is not None
assert result.name == "empty_dict"
+736
View File
@@ -0,0 +1,736 @@
"""Comprehensive tests for set_category_budget overlap handling."""
from datetime import UTC, date, datetime
import pytest
from sqlalchemy.orm import Session
from src.domain_utils import find_category_db, set_category_budget
from src.models import Category, CategoryBudget, CategoryBudgetDb
from src.models.category import CategoryDb, upsert_category
from tests.factories import CategoryFactory
# NOTE: MYPY errors suppressed for test fixtures - see AGENTS.md for type checking notes.
class TestSetCategoryBudgetNoOverlap:
"""Test budgets that don't overlap with existing ones."""
def test_no_overlap_add_new(self, db_session: Session):
"""When new budget has no overlap with existing, add both."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# First budget: Oct 1 - Nov 30
budget1 = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, budget1)
# Second budget: Dec 1 - Dec 31 (no overlap with Oct-Nov range)
budget2 = CategoryBudget(
category=cat,
amount_eur=150.00,
valid_from=date(2026, 12, 1),
valid_to=date(2026, 12, 31),
)
set_category_budget(db_session, budget2)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2
def test_completely_before_no_overlap(self, db_session: Session):
"""New budget completely before existing - should add both."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# Existing budget covering Oct 15 - Dec 30
existing_budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 15),
valid_to=date(2026, 12, 30),
)
set_category_budget(db_session, existing_budget)
# New budget covering Sep 1 - Sep 30 (completely before)
new_budget = CategoryBudget(
category=cat,
amount_eur=50.00,
valid_from=date(2026, 9, 1),
valid_to=date(2026, 9, 30),
)
set_category_budget(db_session, new_budget)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2
class TestSetCategoryBudgetFullOverlap:
"""Test when new budget fully overlaps existing."""
def test_full_overlap_replace(self, db_session: Session):
"""When new budget fully overlaps existing, replace entirely."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# Existing budget covering Oct 1 - Nov 30
existing_budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, existing_budget)
# New budget covering Oct 1 - Nov 30 (same range, different amount)
new_budget = CategoryBudget(
category=cat,
amount_eur=200.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, new_budget)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
# Should have only 1 budget with the new amount (replaced old one)
assert len(budgets) == 1
assert budgets[0].amount_eur == 200.00
class TestSetCategoryBudgetPartialOverlapStart:
"""Test partial overlap where new budget overlaps at start."""
def test_partial_overlap_at_start(self, db_session: Session):
"""When new budget overlaps existing from the start, keep end and cut before."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# Existing budget covering Oct 1 - Nov 30
existing_budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, existing_budget)
# New budget covering Oct 1 - Nov 15 (overlaps from start)
new_budget = CategoryBudget(
category=cat,
amount_eur=120.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 15),
)
set_category_budget(db_session, new_budget)
# Should have 2 budgets:
# 1. Oct 1 - Nov 15 (new budget)
# 2. Dec 1 - Dec 30 (remainder from old budget, after overlap end + 1 day)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2
class TestSetCategoryBudgetPartialOverlapEnd:
"""Test partial overlap where new budget overlaps at end."""
def test_partial_overlap_at_end(self, db_session: Session):
"""When new budget overlaps existing at the end, keep start and cut after."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# Existing budget covering Oct 1 - Nov 30
existing_budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, existing_budget)
# New budget covering Oct 15 - Nov 30 (overlaps at end)
new_budget = CategoryBudget(
category=cat,
amount_eur=120.00,
valid_from=date(2026, 10, 15),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, new_budget)
# Should have 2 budgets:
# 1. Oct 1 - Oct 14 (remainder from old budget, before overlap start - 1 day)
# 2. Oct 15 - Nov 30 (new budget)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2
class TestSetCategoryBudgetMiddleOverlap:
"""Test overlap in the middle that splits existing budget."""
def test_complex_overlap_middle(self, db_session: Session):
"""Handle overlap in the middle that splits existing budget into two parts."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# Existing budget covering Oct 1 - Dec 31
existing_budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 12, 31),
)
set_category_budget(db_session, existing_budget)
# New budget covering Oct 15 - Nov 30 (overlaps in middle)
new_budget = CategoryBudget(
category=cat,
amount_eur=150.00,
valid_from=date(2026, 10, 15),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, new_budget)
# Should have 3 budgets:
# 1. Oct 1 - Oct 14 (before new range, original amount)
# 2. Oct 15 - Nov 30 (new budget, new amount)
# 3. Dec 1 - Dec 31 (after new range, original amount)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 3
class TestSetCategoryBudgetUnbounded:
"""Test budgets with unbounded dates."""
def test_unbounded_start(self, db_session: Session):
"""Budget with None valid_from (unbounded start)."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=None, # Unbounded start
valid_to=date(2026, 12, 31),
)
set_category_budget(db_session, budget)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 1
assert budgets[0].valid_from is None
assert budgets[0].valid_to == date(2026, 12, 31)
def test_unbounded_end(self, db_session: Session):
"""Budget with None valid_to (unbounded end)."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=None, # Unbounded end
)
set_category_budget(db_session, budget)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 1
assert budgets[0].valid_from == date(2026, 10, 1)
assert budgets[0].valid_to is None
def test_unbounded_both(self, db_session: Session):
"""Budget with both dates unbounded (None)."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=None, # Unbounded start
valid_to=None, # Unbounded end
)
set_category_budget(db_session, budget)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 1
assert budgets[0].valid_from is None
assert budgets[0].valid_to is None
class TestSetCategoryBudgetWithExistingUnbounded:
"""Test adding bounded budgets to categories with existing unbounded budgets."""
def test_add_bounded_after_unbounded(self, db_session: Session):
"""Add a bounded budget after an unbounded one (no overlap)."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# First, add unbounded budget
unbounded_budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=None,
valid_to=None,
)
set_category_budget(db_session, unbounded_budget)
# Add bounded budget after (Nov 1 - Dec 31)
bounded_budget = CategoryBudget(
category=cat,
amount_eur=150.00,
valid_from=date(2026, 11, 1),
valid_to=date(2026, 12, 31),
)
set_category_budget(db_session, bounded_budget)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2
def test_add_bounded_before_unbounded(self, db_session: Session):
"""Add a bounded budget before an unbounded one (no overlap)."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# First, add unbounded budget
unbounded_budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=None,
valid_to=None,
)
set_category_budget(db_session, unbounded_budget)
# Add bounded budget before (Sep 1 - Oct 31)
bounded_budget = CategoryBudget(
category=cat,
amount_eur=75.00,
valid_from=date(2026, 9, 1),
valid_to=date(2026, 10, 31),
)
set_category_budget(db_session, bounded_budget)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2
class TestSetCategoryBudgetMultipleWithSameAmount:
"""Test multiple budgets with same amount and overlapping dates."""
def test_same_range_different_amount(self, db_session: Session):
"""When two budgets have same range but different amounts, only one should remain."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# First budget
budget1 = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, budget1)
# Same range but different amount
budget2 = CategoryBudget(
category=cat,
amount_eur=200.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, budget2)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 1
assert budgets[0].amount_eur == 200.00
class TestSetCategoryBudgetDateEdgeCases:
"""Test edge cases around date boundaries."""
def test_adjacent_budgets_no_overlap(self, db_session: Session):
"""Two budgets that are adjacent (end of one is start-1 of next) should both exist."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
# Budget 1: Oct 1 - Oct 31
budget1 = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 10, 31),
)
set_category_budget(db_session, budget1)
# Budget 2: Nov 1 - Nov 30 (adjacent to Oct 31)
budget2 = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 11, 1),
valid_to=date(2026, 11, 30),
)
set_category_budget(db_session, budget2)
budgets = list(
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id)
)
assert len(budgets) == 2
def test_shared_endpoint_is_replaced(self, db_session: Session):
"""Inclusive ranges that share a date must not overlap after replacement."""
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
set_category_budget(
db_session,
CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 10, 15),
),
)
set_category_budget(
db_session,
CategoryBudget(
category=cat,
amount_eur=200.00,
valid_from=date(2026, 10, 15),
valid_to=date(2026, 10, 31),
),
)
budgets = (
db_session.query(CategoryBudgetDb)
.filter_by(category_id=cat_db.id)
.order_by(CategoryBudgetDb.valid_from)
.all()
)
assert [(b.amount_eur, b.valid_from, b.valid_to) for b in budgets] == [
(100.00, date(2026, 10, 1), date(2026, 10, 14)),
(200.00, date(2026, 10, 15), date(2026, 10, 31)),
]
class TestSetCategoryBudgetRegressionCases:
def test_new_range_overlapping_multiple_budgets_is_added_once(
self, db_session: Session
):
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
for start, end, amount in (
(date(2026, 10, 1), date(2026, 10, 31), 100.00),
(date(2026, 12, 1), date(2026, 12, 31), 200.00),
):
set_category_budget(
db_session,
CategoryBudget(
category=cat,
amount_eur=amount,
valid_from=start,
valid_to=end,
),
)
set_category_budget(
db_session,
CategoryBudget(
category=cat,
amount_eur=300.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 12, 31),
),
)
budgets = (
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id).all()
)
assert len(budgets) == 1
assert budgets[0].amount_eur == 300.00
assert budgets[0].valid_from == date(2026, 10, 1)
assert budgets[0].valid_to == date(2026, 12, 31)
@pytest.mark.parametrize(
("existing_start", "existing_end", "new_start", "new_end", "expected"),
[
(
date(2026, 10, 1),
date(2026, 12, 31),
None,
date(2026, 11, 15),
[
(100.00, date(2026, 11, 16), date(2026, 12, 31)),
(200.00, None, date(2026, 11, 15)),
],
),
(
date(2026, 10, 1),
date(2026, 12, 31),
date(2026, 11, 15),
None,
[
(100.00, date(2026, 10, 1), date(2026, 11, 14)),
(200.00, date(2026, 11, 15), None),
],
),
(
None,
date(2026, 12, 31),
date(2026, 10, 1),
date(2026, 10, 31),
[
(100.00, None, date(2026, 9, 30)),
(200.00, date(2026, 10, 1), date(2026, 10, 31)),
(100.00, date(2026, 11, 1), date(2026, 12, 31)),
],
),
(
date(2026, 10, 1),
None,
date(2026, 11, 1),
date(2026, 11, 30),
[
(100.00, date(2026, 10, 1), date(2026, 10, 31)),
(200.00, date(2026, 11, 1), date(2026, 11, 30)),
(100.00, date(2026, 12, 1), None),
],
),
],
)
def test_partially_unbounded_ranges_are_split_correctly(
self,
db_session: Session,
existing_start: date | None,
existing_end: date | None,
new_start: date | None,
new_end: date | None,
expected: list[tuple[float, date | None, date | None]],
):
cat = CategoryFactory.build()
cat_db = upsert_category(db_session, cat, do_commit=False)
db_session.commit()
set_category_budget(
db_session,
CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=existing_start,
valid_to=existing_end,
),
)
set_category_budget(
db_session,
CategoryBudget(
category=cat,
amount_eur=200.00,
valid_from=new_start,
valid_to=new_end,
),
)
budgets = (
db_session.query(CategoryBudgetDb).filter_by(category_id=cat_db.id).all()
)
actual = {(float(b.amount_eur), b.valid_from, b.valid_to) for b in budgets}
assert actual == set(expected)
class TestFindCategoryDb:
"""Test find_category_db function."""
def test_find_simple(self, db_session: Session):
"""Find a simple category without parent."""
cat = Category(name="SimpleCategory")
upsert_category(db_session, cat, do_commit=False)
db_session.commit()
result = find_category_db(db_session, cat)
assert result is not None
assert str(result.name) == "SimpleCategory"
def test_find_with_parent_hierarchy(self, db_session: Session):
"""Find a category with parent hierarchy."""
grandfather = Category(name="grandfather")
upsert_category(db_session, grandfather, do_commit=False)
db_session.commit()
father = Category(name="father", parent=grandfather)
upsert_category(db_session, father, do_commit=False)
db_session.commit()
child = Category(name="child", parent=father)
upsert_category(db_session, child, do_commit=False)
db_session.commit()
# Find the child via find_category_db
result = find_category_db(db_session, child)
assert result is not None
assert str(result.parent.name) == "father"
def test_find_nonexistent(self, db_session: Session):
"""Return None for non-existent category."""
# Create a category but don't add it to session (already in memory from fixtures)
result = find_category_db(db_session, Category(name="NonExistent"))
assert result is None
def test_find_reintroduced_category_returns_active_version(
self, db_session: Session
):
category = Category(name="Reintroduced")
historical = CategoryDb(
name=category.name,
valid_to=datetime(2026, 1, 1, tzinfo=UTC),
last_action="retired",
)
db_session.add(historical)
db_session.commit()
active = upsert_category(db_session, category, do_commit=False)
assert active is not None
db_session.commit()
result = find_category_db(db_session, category)
assert result is not None
assert result.id == active.id
set_category_budget(
db_session,
CategoryBudget(category=category, amount_eur=100.00),
)
budget = db_session.query(CategoryBudgetDb).one()
assert budget.category_id == active.id
class TestSetCategoryBudgetErrorHandling:
"""Test error handling in set_category_budget."""
def test_missing_category_raises_error(self, db_session: Session):
"""set_category_budget should raise ValueError for missing category."""
cat = Category(name="NonExistent")
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 10, 31),
)
with pytest.raises(ValueError) as exc_info:
set_category_budget(db_session, budget)
assert "NonExistent" in str(exc_info.value)
def test_missing_parent_raises_error(self, db_session: Session):
"""set_category_budget should raise ValueError for missing parent."""
cat = Category(name="ChildOfMissing")
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 10, 31),
)
with pytest.raises(ValueError) as exc_info:
set_category_budget(db_session, budget)
# The error should mention that the category was not found (including its parent)
assert (
"ChildOfMissing" in str(exc_info.value).lower()
or "not found" in str(exc_info.value).lower()
or "no matching" in str(exc_info.value).lower()
)
class TestSetCategoryBudgetValidation:
"""Test CategoryBudget model validation rules."""
def test_valid_budget_no_dates(self):
"""Budget with None dates is valid."""
cat = CategoryFactory.build()
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
)
assert budget.valid_from is None
assert budget.valid_to is None
def test_valid_budget_with_dates(self):
"""Budget with dates is valid."""
cat = CategoryFactory.build()
budget = CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
assert budget.valid_from == date(2026, 10, 1)
assert budget.valid_to == date(2026, 11, 30)
def test_invalid_date_range_raises(self):
"""Budget with valid_from > valid_to should raise validation error."""
cat = CategoryFactory.build()
with pytest.raises(ValueError) as exc_info:
CategoryBudget(
category=cat,
amount_eur=100.00,
valid_from=date(2026, 11, 30),
valid_to=date(2026, 10, 1),
)
assert "valid_from" in str(exc_info.value) or "valid_to" in str(exc_info.value)
def test_amount_must_be_positive(self):
"""Amount must be greater than zero."""
cat = CategoryFactory.build()
with pytest.raises(ValueError) as exc_info:
CategoryBudget(
category=cat,
amount_eur=-100.00,
valid_from=date(2026, 10, 1),
valid_to=date(2026, 11, 30),
)
assert "greater than" in str(exc_info.value).lower()
+125
View File
@@ -0,0 +1,125 @@
"""Tests for extract_categories.py - extracting unique categories from b4w CSV exports."""
import csv
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))
from scripts.initial_load.categories.extract_categories import extract_unique_categories
def write_csv(path: Path, categories: list[str]) -> None:
with open(path, "w", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=["Id", "AcctId", "Category"])
writer.writeheader()
for i, category in enumerate(categories, 1):
writer.writerow({"Id": str(i), "AcctId": "587713594", "Category": category})
def read_lines(path: Path) -> list[str]:
return path.read_text(encoding="utf-8").splitlines()
def test_extract_unique_categories_basic(tmp_path):
csv_file = tmp_path / "input.csv"
output_file = tmp_path / "categories.txt"
write_csv(csv_file, ["Girokonto:Lebenshaltung", "Girokonto:Sparen"])
result = extract_unique_categories(str(csv_file), str(output_file))
assert result == ["Girokonto:Lebenshaltung", "Girokonto:Sparen"]
assert read_lines(output_file) == ["Girokonto:Lebenshaltung", "Girokonto:Sparen"]
def test_extract_unique_categories_no_duplicates(tmp_path):
csv_file = tmp_path / "input.csv"
output_file = tmp_path / "categories.txt"
write_csv(
csv_file,
["Girokonto:Lebenshaltung", "Girokonto:Sparen", "Girokonto:Lebenshaltung"],
)
extract_unique_categories(str(csv_file), str(output_file))
assert len(read_lines(output_file)) == 2
def test_extract_unique_categories_output_format(tmp_path):
csv_file = tmp_path / "input.csv"
output_file = tmp_path / "categories.txt"
write_csv(
csv_file,
["Girokonto:Lebenshaltung", "Girokonto:Sparen", "Girokonto:Privat"],
)
extract_unique_categories(str(csv_file), str(output_file))
lines = read_lines(output_file)
assert len(lines) == 3
# Plain list: no numbering prefix, no blank lines.
assert all(line == line.strip() and not line[0].isdigit() for line in lines)
def test_extract_unique_categories_german_special_chars(tmp_path):
csv_file = tmp_path / "input.csv"
output_file = tmp_path / "categories.txt"
write_csv(
csv_file,
["Girokonto:Freizeit:Urlaub&Ausflüge", "Girokonto:Privat:Mitgliedsbeiträge"],
)
extract_unique_categories(str(csv_file), str(output_file))
assert read_lines(output_file) == [
"Girokonto:Freizeit:Urlaub&Ausflüge",
"Girokonto:Privat:Mitgliedsbeiträge",
]
def test_extract_unique_categories_handles_empty_csv(tmp_path):
csv_file = tmp_path / "input.csv"
output_file = tmp_path / "categories.txt"
csv_file.write_text("Id,AcctId,Category\n", encoding="utf-8")
extract_unique_categories(str(csv_file), str(output_file))
assert output_file.exists()
assert read_lines(output_file) == []
def test_extract_unique_categories_handles_csv_with_all_empty_categories(tmp_path):
csv_file = tmp_path / "input.csv"
output_file = tmp_path / "categories.txt"
csv_file.write_text(
"Id,AcctId,Category\n1,587713594,\n2,587713594,\n3,587713594,\n",
encoding="utf-8",
)
extract_unique_categories(str(csv_file), str(output_file))
assert read_lines(output_file) == []
def test_extract_unique_categories_rerun_is_idempotent(tmp_path):
csv_file = tmp_path / "input.csv"
output_file = tmp_path / "categories.txt"
write_csv(csv_file, ["Girokonto:Lebenshaltung", "Girokonto:Sparen"])
extract_unique_categories(str(csv_file), str(output_file))
first_run = read_lines(output_file)
# A new category appears in the CSV on the second run.
write_csv(
csv_file, ["Girokonto:Lebenshaltung", "Girokonto:Sparen", "Girokonto:Privat"]
)
extract_unique_categories(str(csv_file), str(output_file))
second_run = read_lines(output_file)
# Existing entries are merged, not duplicated; list stays sorted and plain.
assert second_run == [
"Girokonto:Lebenshaltung",
"Girokonto:Privat",
"Girokonto:Sparen",
]
assert first_run == ["Girokonto:Lebenshaltung", "Girokonto:Sparen"]
+86 -1
View File
@@ -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()
+86 -49
View File
@@ -1,56 +1,93 @@
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():
c = Category.from_str("single")
assert c
assert c.name == "single"
assert c.parent is None
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"))
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 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():
c = Category.from_str("first:second:third")
assert str(c) == "first:second:third"
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_category_repr():
c = Category.from_str("first:second:third")
assert repr(c) == 'Category.from_str("first:second:third")'
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
def test_category_hash_id_computed_field():
"""Test hash_id computed field is present and deterministic."""
c1 = Category.from_str("single")
c2 = Category.from_str("single")
assert hasattr(c1, "hash_id")
assert c1.hash_id == c2.hash_id
def test_category_hash_id_different_inputs():
"""Test that different category hierarchies have different hash_ids."""
c1 = Category.from_str("A:B")
c2 = Category.from_str("A:C")
# Different last level -> different hash_id
assert c1.hash_id != c2.hash_id
def test_category_to_str_single_level():
"""Test to_str() for single-level category."""
c = Category.from_str("single")
assert c.to_str() == "single"
def test_category_to_str_nested():
"""Test to_str() for nested category."""
c = Category.from_str("parent:child:grandchild")
assert c.to_str() == "parent:child:grandchild"
def test_category_str_delegates_to_to_str():
"""Test that __str__ delegates to to_str()."""
c = Category.from_str("A:B:C")
assert str(c) == c.to_str() == "A:B:C"
def test_category_repr_simple():
"""Test repr() for single-level category."""
c = Category.from_str("single")
expected = 'Category.from_str("single")'
assert repr(c) == expected
def test_category_repr_nested():
"""Test repr() for nested category."""
c = Category.from_str("parent:child")
expected = 'Category.from_str("parent:child")'
assert repr(c) == expected
+145
View File
@@ -0,0 +1,145 @@
"""Tests for common model utilities."""
from datetime import date
from src.models.common import ExportBaseModel, Sign, date2str
class TestSignEnum:
"""Test Sign enum from common module."""
def test_sign_str(self):
"""Test Sign.__str__ method."""
assert str(Sign.DEBIT) == "DBIT"
assert str(Sign.CREDIT) == "CRDT"
def test_sign_repr(self):
"""Test Sign.__repr__ method (line 24)."""
repr_debit = repr(Sign.DEBIT)
assert repr_debit == "Sign.DEBIT"
repr_credit = repr(Sign.CREDIT)
assert repr_credit == "Sign.CREDIT"
def test_sign_to_sign(self):
"""Test Sign.to_sign method."""
assert Sign.DEBIT.to_sign() == -1
assert Sign.CREDIT.to_sign() == 1
class TestDate2Str:
"""Test date2str utility function from common module."""
def test_with_date(self):
"""date2str with a valid date."""
test_date = date(2026, 10, 15)
result = date2str(test_date)
assert result == "2026-10-15"
def test_with_none_default(self):
"""date2str with None value (line 40)."""
result = date2str(None)
assert result == "--"
def test_with_none_custom_default(self):
"""date2str with None and custom default."""
result = date2str(None, default="N/A")
assert result == "N/A"
def test_with_date_custom_format(self):
"""date2str with custom format."""
test_date = date(2026, 10, 15)
result = date2str(test_date, date_format="%d.%m.%Y")
assert result == "15.10.2026"
def test_with_none_custom_default_and_format(self):
"""date2str with None, custom default and format."""
result = date2str(None, default="--", date_format="%d.%m.%Y")
assert result == "--"
class TestExportBaseModel:
"""Test ExportBaseModel methods from common module."""
def test_parameters_to_dict_empty(self):
"""parameters_to_dict returns empty dict (line 70)."""
class TestModel(ExportBaseModel):
pass
model = TestModel()
result = model.parameters_to_dict()
assert result == {}
def test_parameters_to_str_no_params(self):
"""parameters_to_str with no parameters."""
class TestModel(ExportBaseModel):
pass
model = TestModel()
result = model.parameters_to_str()
assert result == ""
def test_repr_empty_params(self):
"""__repr__ with empty parameters (line 79)."""
class TestModel(ExportBaseModel):
pass
model = TestModel()
# __repr__ should return string representation with empty params
repr_result = repr(model)
assert "TestModel" in repr_result
def test_str_empty_params(self):
"""__str__ with empty parameters."""
class TestModel(ExportBaseModel):
pass
model = TestModel()
# __str__ should return string representation
str_result = str(model)
assert "TestModel" in str_result
def test_repr_with_params(self):
"""Test repr with actual parameters."""
class ConfigurableModel(ExportBaseModel):
def parameters_to_dict(self) -> dict[str, str]:
return {"key": "value"}
model = ConfigurableModel()
# Test that repr works correctly with params
repr_result = repr(model)
assert "key" in repr_result or "value" in repr_result
def test_str_with_params(self):
"""Test str with actual parameters."""
class ConfigurableModel(ExportBaseModel):
def parameters_to_dict(self) -> dict[str, str]:
return {"name": "test"}
model = ConfigurableModel()
# Test that str works correctly with params
str_result = str(model)
assert "name" in str_result or "test" in str_result
class TestMyNameMethod:
"""Test ExportBaseModel._my_name classmethod."""
def test_my_name_returns_classname(self):
"""_my_name should return the class name."""
class SubModel(ExportBaseModel):
pass
result = SubModel._my_name()
assert result == "SubModel"
+110
View File
@@ -0,0 +1,110 @@
"""Tests for src.models.__init__.py lazy loading and __dir__."""
import pytest
from src import models
class TestLazyLoading:
"""Tests for lazy attribute loading via __getattr__."""
def test_load_b4wexport(self):
"""Test loading B4wExport model."""
from src.models.b4w import B4wExport as DirectB4wExport
b4w = models.B4wExport
assert b4w is DirectB4wExport
def test_load_category(self):
"""Test loading Category model."""
from src.models.category import Category as DirectCategory
cat = models.Category
assert cat is DirectCategory
def test_load_categorybudgetdb(self):
"""Test loading CategoryBudgetDb model."""
from src.models.category import CategoryBudgetDb as DirectCategoryBudgetDb
cbdb = models.CategoryBudgetDb
assert cbdb is DirectCategoryBudgetDb
def test_load_comdirectgiro(self):
"""Test loading ComdirectGiro model."""
from src.models.comdirect import ComdirectGiro as DirectComdirectGiro
giro = models.ComdirectGiro
assert giro is DirectComdirectGiro
def test_load_transaction(self):
"""Test loading Transaction model."""
from src.models.transactions import Transaction as DirectTransaction
tx = models.Transaction
assert tx is DirectTransaction
def test_load_account(self):
"""Test loading Account model."""
from src.models.settings import Account as DirectAccount
acc = models.Account
assert acc is DirectAccount
def test_load_source(self):
"""Test loading Source model."""
from src.models.settings import Source as DirectSource
src = models.Source
assert src is DirectSource
def test_nonexistent_attribute_raises(self):
"""Test that non-existent attributes raise AttributeError."""
with pytest.raises(AttributeError) as exc_info:
# This is intentional: testing that accessing a non-existent attribute raises
_ = models.NonExistentModel # type: ignore[name-defined]
expected_msg = "module 'src.models' has no attribute 'NonExistentModel'"
assert expected_msg in str(exc_info.value)
class TestDirMethod:
"""Tests for __dir__() method."""
def test_dir_returns_sorted_list(self):
"""Test that __dir__() returns a sorted list."""
result = models.__dir__()
assert isinstance(result, list)
# Should be alphabetically sorted
assert result == sorted(result)
def test_dir_includes_all_exports(self):
"""Test that __dir__() includes all exported models."""
result = models.__dir__()
expected_exports = {
"B4wExport",
"Category",
"CategoryBudgetDb",
"ComdirectCreditCard",
"ComdirectDepot",
"ComdirectGiro",
"Account",
"Source",
"Transaction",
"TransactionDb",
}
# All exports should be in dir()
for export_name in expected_exports:
assert export_name in result
def test_dir_includes_globals(self):
"""Test that __dir__() includes module-level globals."""
result = models.__dir__()
assert "_EXPORTS" in result
def test_dir_is_alphabetically_sorted(self):
"""Verify the list is truly sorted alphabetically."""
result = models.__dir__()
# Double-check: if we sort it again, it should be identical
assert result == sorted(result)
+34
View File
@@ -0,0 +1,34 @@
from datetime import date
import faker
from sqlalchemy.orm import Session
from src.models.b4w import B4wDb
from src.models.category import CategoryDb, upsert_category
from tests.factories import CategoryFactory
def _load_initial_categories(session: Session) -> None:
for category in CategoryFactory.build_all():
upsert_category(session, category, do_commit=False)
session.commit()
def test_b4w_populate(populated_db_b4w):
val_dt_min, val_dt_max = date(2026, 1, 1), date(2026, 3, 31)
session = populated_db_b4w(
"b4w_basic",
100,
setup=_load_initial_categories,
ValDt=lambda: faker.Faker().date_between(val_dt_min, val_dt_max),
)
assert session
assert session.query(B4wDb).count() == 100
assert session.query(CategoryDb).count() >= len(CategoryFactory.build_all())
# ``ValDt`` was given as a per-row callable: it must have been resolved
# and persisted for every row.
dates = [d for (d,) in session.query(B4wDb.date_valid).all()]
assert all(val_dt_min <= d <= val_dt_max for d in dates)
# ``Id1`` is assigned per row: one distinct row per id, no dedup collisions.
ids = [i for (i,) in session.query(B4wDb.id_b4w).order_by(B4wDb.id_b4w).all()]
assert ids == list(range(1, 101))
+2 -2
View File
@@ -15,7 +15,7 @@ from src.ingest.transactions import (
from src.models.b4w import B4wDb
from src.models.comdirect import ComdirectGiroDb
from src.models.transactions import Transaction, TransactionDb
from tests.factories import B4wExportFactory
from tests.factories import B4wExportFactory, CategoryFactory
B4W_PATH = Path("tests/test_data/b4w_export_dummy_01.csv")
COMDIRECT_PATH = Path("tests/test_data/comdirect_export_dummy_01.txt")
@@ -108,7 +108,7 @@ def test_consolidate_b4w_signed_amount_and_category(db_session: Session):
export = B4wExportFactory.build(
Amt="50.00",
RmtInf="Supermarket",
Category="Lebenshaltung:Lebensmittel",
Category=CategoryFactory.build().to_str(),
CdtDbtInd="DBIT",
)
staging = upsert_b4w(db_session, export, do_commit=False)
Generated
+85
View File
@@ -142,6 +142,75 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" },
]
[[package]]
name = "coverage"
version = "7.16.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/2f/55/d1eaf3e73781174340a00dc1ba2aee8a65f82fadb18e2797b192b6b3925b/coverage-7.16.2.tar.gz", hash = "sha256:ca64d9f1f384f151b9511bec01126072acd2f313439f8ed015a22d8790aab6fa", size = 971999, upload-time = "2026-09-27T12:29:01.118Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/59/4c/577fc0803dab4155dcf808faffbdd7b159256781c0874a8586e17b81b149/coverage-7.16.2-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:4ee546b9e4872ffa194bf07ac87bfa1202ebb824d0795dc1ef22f175545ca90a", size = 224138, upload-time = "2026-09-27T12:27:05.141Z" },
{ url = "https://files.pythonhosted.org/packages/75/9e/e3785ba3ecba2bd11efc74bfe2801ca4b78c4480b15a375648d809a59da3/coverage-7.16.2-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:a2fac6895eb299a2e52d7bbb8fb3903502b9da8d3f5309ceb16ec40c646b58ee", size = 224445, upload-time = "2026-09-27T12:27:06.805Z" },
{ url = "https://files.pythonhosted.org/packages/f0/d0/963ff22d3fd27117da3b8cc442f5bdc91196f783321e1a8ff0ec43476772/coverage-7.16.2-cp314-cp314-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:57ff3783f99d75a1e81dd56a9737eb5665e6736a5d93258ba596b6dcad8fd05b", size = 256112, upload-time = "2026-09-27T12:27:08.43Z" },
{ url = "https://files.pythonhosted.org/packages/a8/d4/a306940c81c6ae759e82fff27d20b7fdc6896e422b821f51313cce212b6c/coverage-7.16.2-cp314-cp314-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:35f37886699cb9abd29958247d718628d5bc6f39e623dff66a09e546c42a7e03", size = 258727, upload-time = "2026-09-27T12:27:09.927Z" },
{ url = "https://files.pythonhosted.org/packages/b9/a3/d3d99d93b02517087aa05bc0cf2d04d372956b849e5443e059079901429b/coverage-7.16.2-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0fd7a86fdda7cb6d616d178654bd0ad6bc0f3f33c2e478aa598500a1a9e34eda", size = 259909, upload-time = "2026-09-27T12:27:11.55Z" },
{ url = "https://files.pythonhosted.org/packages/08/44/39dd599181726758dd185ae4dc0c0ab3aeabf7ca70e68e145060feeaaa16/coverage-7.16.2-cp314-cp314-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:ac0f3b379c94acc2f7dce5f5f0b24d44fa1cc6a509717ef83dfee07450c2117c", size = 262481, upload-time = "2026-09-27T12:27:13.17Z" },
{ url = "https://files.pythonhosted.org/packages/99/e8/91ee43f6ded411460c359d7e1aebde4d6fd8f00a2e5394182d9d212eb23c/coverage-7.16.2-cp314-cp314-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:7d0732c83746bc24123c581a85d9dd96b70ddb538c9076020aa1a041790361e9", size = 256012, upload-time = "2026-09-27T12:27:14.91Z" },
{ url = "https://files.pythonhosted.org/packages/11/8c/e9499ddc33197bd7eabcb1118ca81756fc874457b324e2b479a4804b2ad2/coverage-7.16.2-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:7b451c68218c150f616bc9649783ec8de76a59792c759b43aa0c9c0466a465e4", size = 257898, upload-time = "2026-09-27T12:27:16.588Z" },
{ url = "https://files.pythonhosted.org/packages/5f/6e/c081cb5991a0afba99f9c4ad6c74a5fce9513a38ddc64e3e6680c6fed9af/coverage-7.16.2-cp314-cp314-musllinux_1_2_i686.whl", hash = "sha256:a56ac4fa5a75c7e182e8f62600cfb4aff43c5ed7356a034f3557659c3bec1d90", size = 255938, upload-time = "2026-09-27T12:27:18.19Z" },
{ url = "https://files.pythonhosted.org/packages/b2/42/1c3d819e8f9b6eb01c2fe90874d67a8882adb9507e0bbb09361ed131ea89/coverage-7.16.2-cp314-cp314-musllinux_1_2_ppc64le.whl", hash = "sha256:4cc4f73aa3fabc36e32046d6cd2971405948d8a903636508a3d3b2f9128b3a95", size = 260368, upload-time = "2026-09-27T12:27:19.903Z" },
{ url = "https://files.pythonhosted.org/packages/19/4f/d70eac07901fd587b6ab05e659b52afe13959992aa5113bf6cce059cc572/coverage-7.16.2-cp314-cp314-musllinux_1_2_riscv64.whl", hash = "sha256:723dcdab91357159b722935b500ee8abc0a66c8c432e1e9fabf4cc7598952de8", size = 255620, upload-time = "2026-09-27T12:27:21.621Z" },
{ url = "https://files.pythonhosted.org/packages/34/5e/6d87af88317d3d9a9b18a9ca1bc1673eb516917f296e579d0d4a55cb3490/coverage-7.16.2-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:5397e21a90dde0e9c6896b77ded8f0be26b66f8b22b33aed41f6043ed95d55e6", size = 257521, upload-time = "2026-09-27T12:27:23.358Z" },
{ url = "https://files.pythonhosted.org/packages/79/bb/90c2641170d2fa1a6757b3f8450ba2740197317b0ddd749e9604b914e886/coverage-7.16.2-cp314-cp314-win32.whl", hash = "sha256:848893e1d361448c113dc2f0913503522a6f7be231d0e38333d2a22d9698a011", size = 226248, upload-time = "2026-09-27T12:27:25.153Z" },
{ url = "https://files.pythonhosted.org/packages/30/08/d8d0478bb02c8eb0ae20a496fc80c40fcf4d3450bd184300d682ba2d28a6/coverage-7.16.2-cp314-cp314-win_amd64.whl", hash = "sha256:5a27b731c171e43dc8b5f32b76a5051dde2ec9b9366c87028f08a7088ebc2c7b", size = 226732, upload-time = "2026-09-27T12:27:26.907Z" },
{ url = "https://files.pythonhosted.org/packages/32/3f/0001da22155b0a8ce063ec0f7e64ecbe17b373f306e7a74435f6d6accb72/coverage-7.16.2-cp314-cp314-win_arm64.whl", hash = "sha256:1c569a9fd25505f1cd6bea90588818f90373ce90e2632e2cacf19ddbd6e14fdb", size = 226645, upload-time = "2026-09-27T12:27:28.588Z" },
{ url = "https://files.pythonhosted.org/packages/d7/85/6d8813aff9b8b8586691a9d33c43c5604f7227622574da7cdc3d91a86861/coverage-7.16.2-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:d93db87adb6b1c1b408dce4763314b55d76a9f589e96783a84ac9e7689e48bdf", size = 224871, upload-time = "2026-09-27T12:27:30.32Z" },
{ url = "https://files.pythonhosted.org/packages/5c/70/444f3a4981ac2cda40fdcf4cc9b56a4e1a33c222abeb33e51ed3e3eb2a6b/coverage-7.16.2-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:aa62c85046473959c13ba9edca9dc90a77d5c1095b1ba313556314d77fe5b036", size = 225123, upload-time = "2026-09-27T12:27:32.33Z" },
{ url = "https://files.pythonhosted.org/packages/d0/c1/980681cd7b33eb66ac835044116ef0a92e11fcc7bdd866cc89d10b1130b9/coverage-7.16.2-cp314-cp314t-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:db76506aa5416081f3e8974ae0f7965c58ada0bb0ef7339ac86099588dbb20d3", size = 265049, upload-time = "2026-09-27T12:27:34.085Z" },
{ url = "https://files.pythonhosted.org/packages/b2/e3/87679875c33bb2191f0f05544a1cc9adcc940fe0c35443a10f2df753dde5/coverage-7.16.2-cp314-cp314t-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:a0f2285329dac10ab08f79cb11f5692c497018e6c7c511f95e6fd63a70b8f831", size = 267717, upload-time = "2026-09-27T12:27:36.025Z" },
{ url = "https://files.pythonhosted.org/packages/76/64/5d372776d6eb523d4e93bafba2253f96984e3b18261c4cc56a50863c6d0d/coverage-7.16.2-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:382d3346d56b0eec1b793d53a4c88799c8053f516aa3a8d7c44315696954bacf", size = 270047, upload-time = "2026-09-27T12:27:37.96Z" },
{ url = "https://files.pythonhosted.org/packages/be/c1/44082ff0cbf9f97d0043f57970a71204097ec7ba606361a9fd2065393669/coverage-7.16.2-cp314-cp314t-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:648352b94507179d82637292e7ae8802508d95f78e2f00a705a50b6c48011681", size = 271360, upload-time = "2026-09-27T12:27:39.766Z" },
{ url = "https://files.pythonhosted.org/packages/b8/17/9a215efe25b5e0ecc87c89dbe525c4a87d14d87c8c0c7316ef140a5f6f3e/coverage-7.16.2-cp314-cp314t-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:fb2bde05838fffae1a1bf75e5d411a6cac3e4e9bb97e6640fed8cd47888b33f0", size = 264736, upload-time = "2026-09-27T12:27:42.072Z" },
{ url = "https://files.pythonhosted.org/packages/a2/da/7f0a31af8e448107d4d32844bd684757f51ea907bc0c68c8fd537b2123ff/coverage-7.16.2-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:6a75180829efb8ae62b4aded25be6ddca1c888d138d2d82e21d93bfbd88f41cb", size = 267600, upload-time = "2026-09-27T12:27:43.85Z" },
{ url = "https://files.pythonhosted.org/packages/dd/a4/3bfecbd3366b775bacdcb3330394d356cf384b5d8f5b2146ac4b14b252b5/coverage-7.16.2-cp314-cp314t-musllinux_1_2_i686.whl", hash = "sha256:99704f73721e23859112072d522076e11c31744fc96b5652e5dd2018aa4359f7", size = 264767, upload-time = "2026-09-27T12:27:45.768Z" },
{ url = "https://files.pythonhosted.org/packages/b8/3f/5d62163732d87e4a0c4710a0eab30f0fd6a2d480112abe2029f014fe8c9d/coverage-7.16.2-cp314-cp314t-musllinux_1_2_ppc64le.whl", hash = "sha256:29309ccc86b7f33df7db12813c299f215bbbc470ed6292d0bedd63ffae1ebf64", size = 269026, upload-time = "2026-09-27T12:27:47.787Z" },
{ url = "https://files.pythonhosted.org/packages/49/4d/8e4579f225426535085a9be371cc75e3b026d058d679b80affbdfb4c3ef0/coverage-7.16.2-cp314-cp314t-musllinux_1_2_riscv64.whl", hash = "sha256:30c1b65d529e46569899fadca59e4a87c1faf2886923f1307ba61e654d4f3c20", size = 264147, upload-time = "2026-09-27T12:27:49.681Z" },
{ url = "https://files.pythonhosted.org/packages/d1/36/ef1f77e2c3f7bb03c2b13b9a2006f88700fdd75535ef158d70049f425c1c/coverage-7.16.2-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:dcf4bc2aab4e16b1c4c0c2005918f23a7dd5d7821ddae82caed9e3342dc2fcce", size = 266376, upload-time = "2026-09-27T12:27:51.551Z" },
{ url = "https://files.pythonhosted.org/packages/be/79/0cb2bf4428830dec971c718c2c841a039c084415c99e67281f5a72841aab/coverage-7.16.2-cp314-cp314t-win32.whl", hash = "sha256:a9cd3de0a5bfe7b0e21ee10e1a14e3d61bf52efc88217ab1d95d6ace6970bd46", size = 226585, upload-time = "2026-09-27T12:27:53.945Z" },
{ url = "https://files.pythonhosted.org/packages/3c/f9/da17121c16667fd84998e972200ae226a41540f6ea4795776c6d99e8976f/coverage-7.16.2-cp314-cp314t-win_amd64.whl", hash = "sha256:611a44e5229a59d7483ce830160e1a0e85f700562c7a5651c7c63fb8f4eb528c", size = 227378, upload-time = "2026-09-27T12:27:55.778Z" },
{ url = "https://files.pythonhosted.org/packages/74/89/01179c62d1b7e6e33bd5001566b02d7f778cf33d3ec1e81e94ca170c517f/coverage-7.16.2-cp314-cp314t-win_arm64.whl", hash = "sha256:22957cef43ce038641de78ba995de7568d2d6a37c6ddbf7fa0fd7d1ae2344d91", size = 227064, upload-time = "2026-09-27T12:27:57.496Z" },
{ url = "https://files.pythonhosted.org/packages/4c/57/52935003c3f627ba6e5203d7179aad32448c10899663a30336aba8e81a2c/coverage-7.16.2-cp315-cp315-macosx_10_15_x86_64.whl", hash = "sha256:414c26dfdb96aac2d570a54e03008f001e32eb2d413705365503648c6bd361d8", size = 224140, upload-time = "2026-09-27T12:27:59.343Z" },
{ url = "https://files.pythonhosted.org/packages/31/38/df472520f3e626524d7e2fc9d6da0afe7895a2f1489d36b48af8ca40bb41/coverage-7.16.2-cp315-cp315-macosx_11_0_arm64.whl", hash = "sha256:00d3eb96e9988c45f50cccd1f1496571ac5c1f91386ac02c4d55516eeda19a24", size = 224449, upload-time = "2026-09-27T12:28:01.299Z" },
{ url = "https://files.pythonhosted.org/packages/0c/aa/3be084d5b82e63ccdad4ed751e4acbae294673573e30481d29f8b7402eec/coverage-7.16.2-cp315-cp315-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:4dbbd1155ca46e6e0b6b89d204428c56ef6a459af21333f365d135a2820e5a09", size = 256060, upload-time = "2026-09-27T12:28:03.185Z" },
{ url = "https://files.pythonhosted.org/packages/de/29/48fca82a7ebf7ff7b2e35019cc9537e7f65e4d2aa1215cc5a8792c989251/coverage-7.16.2-cp315-cp315-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:8fc15cc8d0d06e873c00ef18e1372d605f9aaf3de27d8c24e50782e75bc8b843", size = 259106, upload-time = "2026-09-27T12:28:05.15Z" },
{ url = "https://files.pythonhosted.org/packages/06/3d/b2d5986f2dd53fe201aa1be2e4ab204fa1aed5101e67c0dbbb419b850aee/coverage-7.16.2-cp315-cp315-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:9c6afdd69218202bc1758c9a14b86b8cf1084f37ed2ca143e567a103772b16d1", size = 260674, upload-time = "2026-09-27T12:28:06.868Z" },
{ url = "https://files.pythonhosted.org/packages/ce/7e/b50160be3506ead12e6480d14279af7f0f17627694300a2d1fd2c42d2ff5/coverage-7.16.2-cp315-cp315-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:aba5c63b7afdc749cc9eae943d5b868cba2b261a176378fa1c5a30bc8bc89982", size = 263139, upload-time = "2026-09-27T12:28:08.771Z" },
{ url = "https://files.pythonhosted.org/packages/14/5e/7c805ac9a32606de1399bd7e9bd375aa2f973dc61b12680d9e6403c2e891/coverage-7.16.2-cp315-cp315-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:9174f0af24e5eff248b9dbfe76ec5275a3d19d37edbc2810543f12cf97347a34", size = 256591, upload-time = "2026-09-27T12:28:10.842Z" },
{ url = "https://files.pythonhosted.org/packages/ab/9e/76f1ed129a2daf658a3ea17122824cf2e3b91fea0460d8d3664fc5a61018/coverage-7.16.2-cp315-cp315-musllinux_1_2_aarch64.whl", hash = "sha256:80e9fdb4c3d926b6ba721d4bf7435bdb869c3527ae7803290361d0ab73db13b6", size = 258708, upload-time = "2026-09-27T12:28:12.962Z" },
{ url = "https://files.pythonhosted.org/packages/5a/b7/8d62e75f48b527619239a65294f842d4b7fd02a0839d43ae1de80184e2df/coverage-7.16.2-cp315-cp315-musllinux_1_2_i686.whl", hash = "sha256:7b3bce4a0d05401d70b7d0d5ca783e686bc9d30e81dbd7d980d532609bf809e4", size = 256562, upload-time = "2026-09-27T12:28:14.934Z" },
{ url = "https://files.pythonhosted.org/packages/b8/8d/0a15f95c3afb78e947c52644786ba4bc9de259905687dd720d5e6fae2e76/coverage-7.16.2-cp315-cp315-musllinux_1_2_ppc64le.whl", hash = "sha256:44f21e407b278efdfc1ee5e481e00518bd1d500310a30a5fbf2bcbedfef4aaf0", size = 261077, upload-time = "2026-09-27T12:28:17.215Z" },
{ url = "https://files.pythonhosted.org/packages/25/00/88389987305a47d732866c07c8a500000ab574df9505e3114ac69c8d027f/coverage-7.16.2-cp315-cp315-musllinux_1_2_riscv64.whl", hash = "sha256:59c3926585e1cd1f2190f4b2ac9014de1bbeaf0d5d0587b0dc6b0aa90d17896a", size = 256034, upload-time = "2026-09-27T12:28:19.08Z" },
{ url = "https://files.pythonhosted.org/packages/92/02/34d079d4952ad461bde037d353f9a6e037a7edc45fe0f9ee8781ff73f028/coverage-7.16.2-cp315-cp315-musllinux_1_2_x86_64.whl", hash = "sha256:066429634299e14dd2d511e1e85f8f9cecc500781f6b41907c0dd6f1baea7e63", size = 258034, upload-time = "2026-09-27T12:28:21.242Z" },
{ url = "https://files.pythonhosted.org/packages/f6/d8/3e59a62879285b464ec1b10fd824fbc1af9ce66e842cd39974f80a0becc4/coverage-7.16.2-cp315-cp315-win32.whl", hash = "sha256:893ea9cf86cb8d2546812ac93d973aaf2ee1fb45110a873b014214fd23e3725e", size = 226252, upload-time = "2026-09-27T12:28:23.102Z" },
{ url = "https://files.pythonhosted.org/packages/f4/e1/128026e1b2836e9ad6b219207ba9edf1c5e0088a7869e23088aee7fbbe7a/coverage-7.16.2-cp315-cp315-win_amd64.whl", hash = "sha256:01c6908bc613b420c26c818fe948e1b97dfd041a53c98b01c63bd8321f5c9aae", size = 226728, upload-time = "2026-09-27T12:28:25.21Z" },
{ url = "https://files.pythonhosted.org/packages/a8/f4/c9fa8e7cf525ca7748ac52b0ee89331d13fe09808e45c679830708782e90/coverage-7.16.2-cp315-cp315-win_arm64.whl", hash = "sha256:967d72c835d7a8cf0af99ec813a2d06e3db6df706402f1fe85b31b437645f495", size = 226648, upload-time = "2026-09-27T12:28:27.136Z" },
{ url = "https://files.pythonhosted.org/packages/a2/13/e96b045447a856666f36f9c653e2a80bdaa732aaaf72412b19aa2c26a473/coverage-7.16.2-cp315-cp315t-macosx_10_15_x86_64.whl", hash = "sha256:98d9c97f51b334b0adce7b964442a9af33c1a00c6ac856984cc5dc8d18f81c75", size = 224858, upload-time = "2026-09-27T12:28:29.169Z" },
{ url = "https://files.pythonhosted.org/packages/23/90/087f6ad1bd3df059632ca3407a4e6552ed1053ee35354de0a771acf35423/coverage-7.16.2-cp315-cp315t-macosx_11_0_arm64.whl", hash = "sha256:3e861f1071dcc2fec1e88bef0920f6b1eaa66a143555b4f8ab79ba2b0f30ef55", size = 225139, upload-time = "2026-09-27T12:28:31.131Z" },
{ url = "https://files.pythonhosted.org/packages/7e/8e/285dcef0184358044e7cbcd810a1bdc9566bc620f54702d605477155df4a/coverage-7.16.2-cp315-cp315t-manylinux1_i686.manylinux_2_28_i686.manylinux_2_5_i686.whl", hash = "sha256:fb9d92ecfe2d5b494367c67f7446f8b75b68d8d0c8cf3bc3e6997478be25d9e2", size = 265090, upload-time = "2026-09-27T12:28:33.04Z" },
{ url = "https://files.pythonhosted.org/packages/06/b2/cc83f3a6e5789a4e89059c69555bc641c2efcde568405a1c06fc702951ab/coverage-7.16.2-cp315-cp315t-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:eb57acff4a74246ae513c142d4b36e18c389c3aed8661914a53f7cd0071031b2", size = 268254, upload-time = "2026-09-27T12:28:35.135Z" },
{ url = "https://files.pythonhosted.org/packages/ac/41/f548c19530f5d66ac6e3c92bbcbc49da7261de3a458b9f3e54a3efb1a0b2/coverage-7.16.2-cp315-cp315t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:444889f7f66b74e4455c0a97e0e166dd41177f1dca8c0239a47cff25e05ba7e1", size = 270695, upload-time = "2026-09-27T12:28:36.959Z" },
{ url = "https://files.pythonhosted.org/packages/94/61/4dc27cf82ef96434d2874110ad0cc10ea4621025705dc5049862bd3bd181/coverage-7.16.2-cp315-cp315t-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:a740ea6f083c6db7b926534d159508f80ba275ab35e722522de0d18d0f56e55f", size = 271853, upload-time = "2026-09-27T12:28:38.821Z" },
{ url = "https://files.pythonhosted.org/packages/38/29/bf8072b1b8bd5f2de8b21460a404460b1a2b97e80a9464c78ec0271f6199/coverage-7.16.2-cp315-cp315t-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:8e209591f7c41ae4a9171335cf6156afda0b21de73b02f73f5aa95b2d5fbb08d", size = 265644, upload-time = "2026-09-27T12:28:40.815Z" },
{ url = "https://files.pythonhosted.org/packages/7c/2f/0aecb8721be5cdeb8afd9d6d9f6b463f074e4d8d37f00f4c42442522709f/coverage-7.16.2-cp315-cp315t-musllinux_1_2_aarch64.whl", hash = "sha256:396bb16e04ce04efbb3df91456ae4e3da918e69ecdf67fb711b0a0fdf35ccce0", size = 268530, upload-time = "2026-09-27T12:28:42.725Z" },
{ url = "https://files.pythonhosted.org/packages/ab/0b/92b4b7628268ee711249958e68fc0328779bd3d9a7ab4715379465aedb84/coverage-7.16.2-cp315-cp315t-musllinux_1_2_i686.whl", hash = "sha256:9cdf19874e0d247f32f03609200370343c3c7aa260b191d8c2bb251d36198283", size = 265154, upload-time = "2026-09-27T12:28:44.684Z" },
{ url = "https://files.pythonhosted.org/packages/7b/d9/41c95c1ab29b3dcd357cd1227181d1c98185632aca41ce670ce671b23a43/coverage-7.16.2-cp315-cp315t-musllinux_1_2_ppc64le.whl", hash = "sha256:fd3d72233eb8b48acc94fa57d44e2d32ce8e7abed02882ccb6d855ccc4ed33ec", size = 269799, upload-time = "2026-09-27T12:28:46.672Z" },
{ url = "https://files.pythonhosted.org/packages/80/07/ebeb259aa5362b033a137b86d7274ff4b109d59be8cc9913889b783bf75a/coverage-7.16.2-cp315-cp315t-musllinux_1_2_riscv64.whl", hash = "sha256:bb4ffe96aa663cee727659db5a2afeb38c95f8677b747d447b90d6d4874ea2c5", size = 265196, upload-time = "2026-09-27T12:28:48.996Z" },
{ url = "https://files.pythonhosted.org/packages/b2/18/8437620f90d023680a072eee02f968055f3658bbfb7d386d0ea34cfb7f30/coverage-7.16.2-cp315-cp315t-musllinux_1_2_x86_64.whl", hash = "sha256:dba2edfb054f6d4a08df9d1637c39a5aa3865bca6617c13c86be21e45658a59c", size = 267266, upload-time = "2026-09-27T12:28:51.361Z" },
{ url = "https://files.pythonhosted.org/packages/28/6c/f08e8ee4293e6434035424180bef4d45e028e8ecc006c61bf9453e74405e/coverage-7.16.2-cp315-cp315t-win32.whl", hash = "sha256:251aed777c47c77aba047096d4542889db089227655711dfc2b9c54ef0e15e35", size = 226583, upload-time = "2026-09-27T12:28:53.33Z" },
{ url = "https://files.pythonhosted.org/packages/f7/fd/3f939c2847f4a72c20cff8b1ac33da78ea91a2d38d9b43336e60db719103/coverage-7.16.2-cp315-cp315t-win_amd64.whl", hash = "sha256:2aca0bdfa9e91621d5b09d815357bf63def4fc0e9cb66da67bf2cf93f3b1a6f5", size = 227374, upload-time = "2026-09-27T12:28:55.158Z" },
{ url = "https://files.pythonhosted.org/packages/5a/35/b98cdc354c952402132e675a87f2cc3227fb68f959c84aaa491fbe15933d/coverage-7.16.2-cp315-cp315t-win_arm64.whl", hash = "sha256:b88841e654f09732804809e435b3e005a929ffd9998b872b7b213957b8759cb8", size = 227064, upload-time = "2026-09-27T12:28:57.075Z" },
{ url = "https://files.pythonhosted.org/packages/3f/0c/7a64e1ac90541a8edf50daef0914848011fb057a5bf55284a4811e21939a/coverage-7.16.2-py3-none-any.whl", hash = "sha256:11d28e9123a9156cb405d8d27b44256c9a58fb5decc2073a8f17862057e3aa0f", size = 215754, upload-time = "2026-09-27T12:28:59.075Z" },
]
[[package]]
name = "debugpy"
version = "1.8.21"
@@ -189,6 +258,7 @@ dependencies = [
{ name = "mypy" },
{ name = "pydantic" },
{ name = "pytest" },
{ name = "pytest-cov" },
{ name = "python-dotenv" },
{ name = "sqlalchemy" },
]
@@ -210,6 +280,7 @@ requires-dist = [
{ name = "mypy", specifier = ">=2.4.0" },
{ name = "pydantic", specifier = ">=2.13.5" },
{ name = "pytest", specifier = ">=9.1.1" },
{ name = "pytest-cov", specifier = ">=7.1.0" },
{ name = "python-dotenv", specifier = ">=1.2.3" },
{ name = "sqlalchemy", specifier = ">=2.0.52" },
]
@@ -564,6 +635,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" },
]
[[package]]
name = "pytest-cov"
version = "7.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "coverage" },
{ name = "pluggy" },
{ name = "pytest" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b1/51/a849f96e117386044471c8ec2bd6cfebacda285da9525c9106aeb28da671/pytest_cov-7.1.0.tar.gz", hash = "sha256:30674f2b5f6351aa09702a9c8c364f6a01c27aae0c1366ae8016160d1efc56b2", size = 55592, upload-time = "2026-03-21T20:11:16.284Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/9d/7a/d968e294073affff457b041c2be9868a40c1c71f4a35fcc1e45e5493067b/pytest_cov-7.1.0-py3-none-any.whl", hash = "sha256:a0461110b7865f9a271aa1b51e516c9a95de9d696734a2f71e3e78f46e1d4678", size = 22876, upload-time = "2026-03-21T20:11:14.438Z" },
]
[[package]]
name = "python-dotenv"
version = "1.2.3"