from django.contrib import messages from django.core.paginator import Paginator from django.db import DatabaseError, connection from django.http import JsonResponse from django.shortcuts import get_object_or_404, redirect, render from django.views.decorators.http import require_POST from finance.models import Account, Category, Payee, Transaction from finance.queries import QueryInputError, TransactionFilters, filtered_transactions from finance.services import ( Actor, DomainError, annotate_transaction, archive, create_account, create_category, create_payee, record_transaction, ) from .forms import AccountForm, AnnotationForm, CategoryForm, ManualTransactionForm, PayeeForm LOCAL_ACTOR = Actor(kind="HUMAN", identifier="local-owner") DOMAIN_MESSAGES = { "archived_reference": "That reference is archived and cannot be selected.", "category_kind_mismatch": "The category does not match the transaction kind.", "conflict": "A record with these values already exists.", "invalid_amount": "Enter a valid amount with at most two decimal places.", "invalid_name": "Enter a name.", "not_found": "The requested record was not found.", "validation_error": "The values do not satisfy the financial rules.", "zero_amount": "The amount cannot be zero.", } UNKNOWN_DOMAIN_MESSAGE = "Unable to save this change." def health(request): try: with connection.cursor() as cursor: cursor.execute("SELECT 1") cursor.fetchone() except DatabaseError: return JsonResponse({"status": "unavailable"}, status=503) return JsonResponse({"status": "ok"}) def home(request): return transaction_list(request) def transaction_list(request): filter_errors = {} try: filters = TransactionFilters.from_mapping(request.GET) rows = filtered_transactions(filters) except QueryInputError as exc: filters = TransactionFilters() rows = Transaction.objects.none() filter_errors = {field: _query_message(code) for field, code in exc.errors.items()} page = Paginator(rows, 50).get_page(request.GET.get("page")) return render(request, "core/transactions/list.html", { "page": page, "filters": filters, "filter_errors": filter_errors, "accounts": Account.objects.order_by("name"), "categories": Category.objects.order_by("name"), "payees": Payee.objects.order_by("name"), "kinds": Transaction.Kind.choices, }) def transaction_detail(request, transaction_id): row = get_object_or_404( Transaction.objects.select_related( "account", "payee", "category", "statement", "refund_of", "installment_plan" ).prefetch_related("source_records"), pk=transaction_id, ) return render(request, "core/transactions/detail.html", {"transaction": row}) def transaction_create(request): form = ManualTransactionForm(request.POST or None) if request.method == "POST" and form.is_valid(): values = form.cleaned_data try: row = record_transaction( account_id=values["account"], amount=values["amount"], posting_date=values["posting_date"], occurrence_date=values["occurrence_date"], kind=values["kind"], description=values["description"], payee_id=values["payee"], category_id=values["category"], actor=LOCAL_ACTOR, ) except DomainError as exc: form.add_error(None, _domain_message(exc)) else: messages.success(request, "Transaction recorded.") return redirect("transaction-detail", transaction_id=row.pk) return render(request, "core/transactions/form.html", {"form": form, "title": "New transaction"}) def transaction_annotate(request, transaction_id): row = get_object_or_404(Transaction.objects.select_related("payee", "category"), pk=transaction_id) form = AnnotationForm(request.POST or None, transaction=row) if request.method == "POST" and form.is_valid(): changes = form.changes() if not changes: return redirect("transaction-detail", transaction_id=row.pk) try: annotate_transaction(transaction_id=row.pk, actor=LOCAL_ACTOR, **changes) except DomainError as exc: form.add_error(None, _domain_message(exc)) else: messages.success(request, "Transaction annotations updated.") return redirect("transaction-detail", transaction_id=row.pk) return render(request, "core/transactions/form.html", { "form": form, "title": "Edit transaction annotations", "transaction": row, }) def account_list(request): return _reference_list(request, Account, "Accounts", "account-create", "account-archive") def category_list(request): return _reference_list(request, Category, "Categories", "category-create", "category-archive") def payee_list(request): return _reference_list(request, Payee, "Payees", "payee-create", "payee-archive") def account_create(request): return _reference_create(request, AccountForm, create_account, "New account", "account-list") def category_create(request): return _reference_create(request, CategoryForm, create_category, "New category", "category-list") def payee_create(request): return _reference_create(request, PayeeForm, create_payee, "New payee", "payee-list") @require_POST def account_archive(request, reference_id): return _reference_archive(request, Account, reference_id, "account-list") @require_POST def category_archive(request, reference_id): return _reference_archive(request, Category, reference_id, "category-list") @require_POST def payee_archive(request, reference_id): return _reference_archive(request, Payee, reference_id, "payee-list") def _reference_list(request, model, title, create_url, archive_url): return render(request, "core/references/list.html", { "title": title, "rows": model.objects.order_by("archived", "name"), "create_url": create_url, "archive_url": archive_url, }) def _reference_create(request, form_class, service, title, redirect_name): form = form_class(request.POST or None) if request.method == "POST" and form.is_valid(): try: service(actor=LOCAL_ACTOR, **form.cleaned_data) except DomainError as exc: form.add_error(None, _domain_message(exc)) else: messages.success(request, f"{title.removeprefix('New ')} created.") return redirect(redirect_name) return render(request, "core/references/form.html", {"form": form, "title": title}) def _reference_archive(request, model, reference_id, redirect_name): item = get_object_or_404(model, pk=reference_id) try: archive(item, actor=LOCAL_ACTOR) except DomainError as exc: messages.error(request, _domain_message(exc)) else: messages.success(request, "Reference archived.") return redirect(redirect_name) def _domain_message(error): return DOMAIN_MESSAGES.get(error.code, UNKNOWN_DOMAIN_MESSAGE) def _query_message(code): return { "invalid_date": "Enter a valid date.", "before_date_from": "The end date must be on or after the start date.", "invalid_uuid": "Select a valid reference.", "invalid_choice": "Select a valid choice.", "invalid_currency": "Enter a three-letter uppercase currency code.", }.get(code, "Invalid filter.")