You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
82 lines
4.2 KiB
82 lines
4.2 KiB
from finance.models import Account, Payee, Category, CategoryGroup, CardStatement, InstallmentPlan, Transaction, SourceRecord |
|
from .base import DomainError, identity, lock, money, save, write |
|
|
|
|
|
def validate_category(kind, category): |
|
if category is None: |
|
return |
|
expected = {"INCOME": "INCOME", "EXPENSE": "EXPENSE", "REFUND": "EXPENSE"}.get(kind) |
|
if expected != category.kind: |
|
raise DomainError("category_kind_mismatch") |
|
|
|
|
|
def lock_category(identifier): |
|
category = lock(Category, identifier, active=True) |
|
if category.group_id is not None: |
|
lock(CategoryGroup, category.group_id, active=True) |
|
return category |
|
|
|
|
|
def record(*, account_id, amount, posting_date, kind, description="", occurrence_date=None, |
|
payee_id=None, category_id=None, statement_id=None, refund_of_id=None, |
|
installment_plan_id=None, installment_number=None): |
|
amount = money(amount) |
|
account = lock(Account, account_id, active=True) |
|
payee = lock(Payee, payee_id, active=True) if payee_id is not None else None |
|
category = lock_category(category_id) if category_id is not None else None |
|
validate_category(kind, category) |
|
statement = lock(CardStatement, statement_id) if statement_id is not None else None |
|
if statement and (account.kind != "CREDIT_CARD" or statement.account_id != account.pk): |
|
raise DomainError("statement_account_mismatch") |
|
plan = lock(InstallmentPlan, installment_plan_id) if installment_plan_id is not None else None |
|
if plan: |
|
if kind != "EXPENSE" or plan.account_id != account.pk: |
|
raise DomainError("installment_account_or_kind_mismatch") |
|
if not isinstance(installment_number, int) or isinstance(installment_number, bool) or not 1 <= installment_number <= plan.total_installments: |
|
raise DomainError("invalid_installment_number") |
|
elif installment_number is not None: |
|
raise DomainError("missing_installment_plan") |
|
original = lock(Transaction, refund_of_id) if refund_of_id is not None else None |
|
if original: |
|
if kind != "REFUND" or original.kind != "EXPENSE" or original.account.currency != account.currency: |
|
raise DomainError("invalid_refund_reference") |
|
obj = Transaction(account=account, amount=amount, posting_date=posting_date, kind=kind, |
|
description=description, occurrence_date=occurrence_date, payee=payee, |
|
category=category, statement=statement, refund_of=original, |
|
installment_plan=plan, installment_number=installment_number) |
|
return save(obj) |
|
|
|
|
|
@write("record_transaction") |
|
def record_transaction(**kwargs): |
|
return record(**kwargs), sorted(key.removesuffix("_id") for key in kwargs) |
|
|
|
|
|
@write("annotate_transaction") |
|
def annotate_transaction(*, transaction_id, **changes): |
|
if not changes or set(changes) - {"description", "payee_id", "category_id"}: |
|
raise DomainError("immutable_transaction") |
|
# Reference rows precede transaction rows in every write path. This avoids |
|
# a cycle with refund posting (category -> original transaction). |
|
references = {} |
|
for field, model in (("payee_id", Payee), ("category_id", Category)): |
|
if field in changes: |
|
value = changes[field] |
|
references[field] = (lock_category(value) if model is Category else lock(model, value, active=True)) if value is not None else None |
|
obj = lock(Transaction, transaction_id) |
|
if "description" in changes: |
|
obj.description = changes["description"] |
|
if "payee_id" in changes: |
|
obj.payee = references["payee_id"] |
|
if "category_id" in changes: |
|
obj.category = references["category_id"] |
|
validate_category(obj.kind, obj.category) |
|
return save(obj), sorted(key.removesuffix("_id") for key in changes) |
|
|
|
|
|
@write("attach_source") |
|
def attach_source(*, transaction_id, document_hash, source_format, parser_version, item_locator): |
|
obj = lock(Transaction, transaction_id) |
|
source = save(SourceRecord(transaction=obj, document_hash=document_hash, source_format=source_format, |
|
parser_version=parser_version, item_locator=item_locator)) |
|
return source, ["transaction", "document_hash", "source_format", "parser_version", "item_locator"]
|
|
|