103 lines
3.6 KiB
Python
103 lines
3.6 KiB
Python
"""Commands and queries for categories and transactions.
|
|
|
|
CQRS: the ``add_*``/``delete_*`` methods are commands (state changes);
|
|
``list_*`` methods are queries (read-only). Neither side leaks the other.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
from ..errors import ConflictError, NotFoundError, ValidationError
|
|
from ..money import Money
|
|
from ..models import Category, Transaction, clean_name, clean_note, validate_kind
|
|
from ..repository import Repository
|
|
|
|
__all__ = ["LedgerService"]
|
|
|
|
|
|
class LedgerService:
|
|
"""Owns category and transaction rules (R5-R10, R12)."""
|
|
|
|
def __init__(self, repo: Repository) -> None:
|
|
self._repo = repo
|
|
|
|
# -- commands ----------------------------------------------------------
|
|
|
|
def add_category(self, name: str, kind: str) -> Category:
|
|
clean = clean_name(name, "category")
|
|
validate_kind(kind)
|
|
if self._repo.find_category(clean) is not None:
|
|
raise ConflictError(f"category {clean!r} already exists")
|
|
self._repo.create_category(clean, kind)
|
|
return Category(name=clean, kind=kind)
|
|
|
|
def delete_category(self, name: str) -> None:
|
|
clean = clean_name(name, "category")
|
|
ref = self._repo.find_category(clean)
|
|
if ref is None:
|
|
raise NotFoundError(f"unknown category {clean!r}")
|
|
category_id, stored_name, _kind = ref
|
|
if self._repo.category_in_use(category_id):
|
|
raise ConflictError(
|
|
f"category {stored_name!r} has transactions or a budget"
|
|
)
|
|
self._repo.delete_category(category_id)
|
|
|
|
def add_transaction(
|
|
self, amount: Money, category: str, day: date, note: str | None = None
|
|
) -> Transaction:
|
|
if not amount.is_positive:
|
|
raise ValidationError("amount must be greater than zero")
|
|
ref = self._repo.find_category(category)
|
|
if ref is None:
|
|
raise NotFoundError(f"unknown category {category!r}")
|
|
category_id, stored_name, kind = ref
|
|
cleaned_note = clean_note(note)
|
|
tx_id = self._repo.insert_transaction(
|
|
amount.cents, category_id, day.isoformat(), cleaned_note
|
|
)
|
|
return Transaction(
|
|
id=tx_id,
|
|
amount=amount,
|
|
category=stored_name,
|
|
kind=kind,
|
|
date=day,
|
|
note=cleaned_note,
|
|
)
|
|
|
|
def delete_transaction(self, tx_id: int) -> None:
|
|
if not self._repo.transaction_exists(tx_id):
|
|
raise NotFoundError(f"unknown transaction #{tx_id}")
|
|
self._repo.delete_transaction(tx_id)
|
|
|
|
# -- queries -----------------------------------------------------------
|
|
|
|
def list_categories(self) -> list[Category]:
|
|
return [Category(name=name, kind=kind) for name, kind in self._repo.list_categories()]
|
|
|
|
def list_transactions(
|
|
self,
|
|
start: str | None,
|
|
end: str | None,
|
|
category: str | None = None,
|
|
) -> list[Transaction]:
|
|
category_id = None
|
|
if category is not None:
|
|
ref = self._repo.find_category(category)
|
|
if ref is None:
|
|
raise NotFoundError(f"unknown category {category!r}")
|
|
category_id = ref[0]
|
|
rows = self._repo.list_transactions(start, end, category_id)
|
|
return [
|
|
Transaction(
|
|
id=row["id"],
|
|
amount=Money(cents=row["amount_cents"]),
|
|
category=row["category"],
|
|
kind=row["kind"],
|
|
date=date.fromisoformat(row["date"]),
|
|
note=row["note"],
|
|
)
|
|
for row in rows
|
|
]
|