from dataclasses import dataclass from datetime import date import re from typing import Mapping from uuid import UUID from django.db.models import Q, QuerySet from finance.models import Account, Category, Payee, Transaction _CURRENCY_RE = re.compile(r"[A-Z]{3}", re.ASCII) class QueryInputError(ValueError): def __init__(self, errors: dict[str, str]): super().__init__("Invalid transaction filters") self.errors = errors @dataclass(frozen=True) class TransactionFilters: date_from: date | None = None date_to: date | None = None account_id: UUID | None = None kind: str | None = None payee_id: UUID | None = None category_id: UUID | None = None currency: str | None = None uncategorized: bool | None = None @classmethod def from_mapping(cls, mapping: Mapping[str, object]) -> "TransactionFilters": errors: dict[str, str] = {} date_from = _parse_date(mapping.get("date_from"), "date_from", errors) date_to = _parse_date(mapping.get("date_to"), "date_to", errors) account_id = _parse_uuid(mapping.get("account"), "account", errors) payee_id = _parse_uuid(mapping.get("payee"), "payee", errors) category_id = _parse_uuid(mapping.get("category"), "category", errors) kind = _optional_string(mapping.get("kind")) if kind is not None and kind not in Transaction.Kind.values: errors["kind"] = "invalid_choice" currency = _optional_string(mapping.get("currency")) if currency is not None and _CURRENCY_RE.fullmatch(currency) is None: errors["currency"] = "invalid_currency" uncategorized_value = _optional_string(mapping.get("uncategorized")) uncategorized = None if uncategorized_value == "yes": uncategorized = True elif uncategorized_value == "no": uncategorized = False elif uncategorized_value is not None: errors["uncategorized"] = "invalid_choice" if date_from is not None and date_to is not None and date_to < date_from: errors["date_to"] = "before_date_from" if errors: raise QueryInputError(errors) return cls( date_from=date_from, date_to=date_to, account_id=account_id, kind=kind, payee_id=payee_id, category_id=category_id, currency=currency, uncategorized=uncategorized, ) def filtered_transactions(filters: TransactionFilters) -> QuerySet[Transaction]: rows = Transaction.objects.select_related( "account", "payee", "category", "category__group", "statement", "refund_of", "installment_plan" ) if filters.date_from is not None: rows = rows.filter(posting_date__gte=filters.date_from) if filters.date_to is not None: rows = rows.filter(posting_date__lte=filters.date_to) if filters.account_id is not None: rows = rows.filter(account_id=filters.account_id) if filters.kind is not None: rows = rows.filter(kind=filters.kind) if filters.payee_id is not None: rows = rows.filter(payee_id=filters.payee_id) if filters.category_id is not None: rows = rows.filter(category_id=filters.category_id) if filters.currency is not None: rows = rows.filter(account__currency=filters.currency) if filters.uncategorized is not None: rows = rows.filter(category__isnull=filters.uncategorized) return rows.order_by("-posting_date", "-id") def listed_accounts() -> QuerySet[Account]: return Account.objects.order_by("archived", "name", "id") def listed_categories() -> QuerySet[Category]: return Category.objects.select_related("group").order_by("group__name", "archived", "name", "id") def listed_payees() -> QuerySet[Payee]: return Payee.objects.order_by("archived", "name", "id") def active_accounts() -> QuerySet[Account]: return Account.objects.filter(archived=False).order_by("name", "id") def active_categories() -> QuerySet[Category]: return Category.objects.filter(archived=False).filter(Q(group__isnull=True) | Q(group__archived=False)).select_related("group").order_by("group__name", "name", "id") def category_label(category: Category) -> str: return f"{category.group.name} ยท {category.name}" if category.group_id else category.name def active_payees() -> QuerySet[Payee]: return Payee.objects.filter(archived=False).order_by("name", "id") def _optional_string(value: object) -> str | None: if value is None: return None text = str(value).strip() return text or None def _parse_date(value: object, field: str, errors: dict[str, str]) -> date | None: text = _optional_string(value) if text is None: return None try: return date.fromisoformat(text) except ValueError: errors[field] = "invalid_date" return None def _parse_uuid(value: object, field: str, errors: dict[str, str]) -> UUID | None: text = _optional_string(value) if text is None: return None try: return UUID(text) except (ValueError, AttributeError): errors[field] = "invalid_uuid" return None