57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
"""Commands and queries for monthly budgets (R2, R7)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from ..errors import NotFoundError, ValidationError
|
|
from ..money import Money
|
|
from ..models import INCOME, Budget, clean_name
|
|
from ..repository import Repository
|
|
|
|
__all__ = ["BudgetService"]
|
|
|
|
|
|
class BudgetService:
|
|
"""Owns budget rules: positivity, expense-only, replace/delete semantics."""
|
|
|
|
def __init__(self, repo: Repository) -> None:
|
|
self._repo = repo
|
|
|
|
# -- commands ----------------------------------------------------------
|
|
|
|
def set_budget(self, category: str, amount: Money) -> Budget:
|
|
clean = clean_name(category, "category")
|
|
if not amount.is_positive:
|
|
raise ValidationError("budget must be greater than zero")
|
|
ref = self._repo.find_category(clean)
|
|
if ref is None:
|
|
raise NotFoundError(f"unknown category {clean!r}")
|
|
category_id, stored_name, kind = ref
|
|
if kind == INCOME:
|
|
raise ValidationError(
|
|
f"budgets apply to expense categories only "
|
|
f"({stored_name!r} is income)"
|
|
)
|
|
self._repo.set_budget(category_id, amount.cents)
|
|
return Budget(category=stored_name, amount=amount)
|
|
|
|
def delete_budget(self, category: str) -> Budget:
|
|
"""Remove the budget for *category*; returns the deleted budget."""
|
|
clean = clean_name(category, "category")
|
|
ref = self._repo.find_category(clean)
|
|
if ref is None:
|
|
raise NotFoundError(f"unknown category {clean!r}")
|
|
category_id, stored_name, _kind = ref
|
|
cents = self._repo.get_budget(category_id)
|
|
if cents is None:
|
|
raise NotFoundError(f"no budget set for {stored_name!r}")
|
|
self._repo.delete_budget(category_id)
|
|
return Budget(category=stored_name, amount=Money(cents=cents))
|
|
|
|
# -- queries -----------------------------------------------------------
|
|
|
|
def list_budgets(self) -> list[Budget]:
|
|
return [
|
|
Budget(category=row["category"], amount=Money(cents=row["amount_cents"]))
|
|
for row in self._repo.list_budgets()
|
|
]
|