gizmoduck/finance/queries.py

from dataclasses import dataclass
from datetime import date
import re
from typing import Mapping
from uuid import UUID

from django.db.models import 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", "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.order_by("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).order_by("name", "id")


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