"""Persistence adapter for pricing and correction value objects. ops-contract: the finance export imports `adjustments.persist` by path. Fold it into another module if you like, but leave this path resolving to these names. """ from __future__ import annotations from pricing.models import ( Account as DomainAccount, Cap, Credit, Invoice as DomainInvoice, Promotion, Tax, ) from adjustments.models import ( Account, AdjustmentRecord, BillingEntry, Invoice, LineItem, ReceivableEntry, RunMembership, RunSummaryEntry, StatementRun, ) def _rule(rule: object) -> dict[str, object]: if isinstance(rule, Promotion): return { "type": "promotion", "key": rule.key, "percent": str(rule.percent), "kind": rule.kind, "applies_to": list(rule.applies_to_usage_keys or ()), } if isinstance(rule, Credit): return {"type": "credit", "key": rule.key, "amount": str(rule.amount)} if isinstance(rule, Cap): return {"type": "cap", "key": rule.key, "amount": str(rule.amount)} if isinstance(rule, Tax): return {"type": "tax", "key": rule.key, "rate": str(rule.rate)} raise ValueError(f"unknown pricing rule: {rule!r}") def base_state(invoice: DomainInvoice) -> dict[str, object]: """Serialize the immutable inputs from which an invoice was priced.""" return { "usage": [ { "key": line.key, "description": line.description, "quantity": str(line.quantity), } for line in invoice.usage ], "effective_from": invoice.rule_set.effective_from, "usage_tiers": [ { "up_to": str(tier.up_to) if tier.up_to is not None else None, "unit_price": str(tier.unit_price), } for tier in invoice.rule_set.usage_tiers ], "rules": [_rule(rule) for rule in invoice.rule_set.rules], } def store_invoice( account: Account, invoice: DomainInvoice, *, issued: bool | None = None ) -> Invoice: """Store a priced invoice and its immutable pricing snapshot.""" row, _ = Invoice.objects.update_or_create( invoice_id=invoice.invoice_id, defaults={ "account": account, "period_key": invoice.period.key, "period_starts_on": invoice.period.starts_on, "period_ends_on": invoice.period.ends_on, "rule_set_key": invoice.rule_set.key, "total": invoice.total, "issued_total": invoice.issued_total, "closed": invoice.closed, "issued": invoice.closed if issued is None else issued, "revision_of": invoice.revision_of, "selected_exclusive_promotion": ( invoice.selected_exclusive_promotion ), "non_selected_exclusive_promotions": list( invoice.non_selected_exclusive_promotions ), "base_state": base_state(invoice), }, ) row.line_items.all().delete() LineItem.objects.bulk_create( [ LineItem( invoice=row, usage_key=item.usage_key, description=item.description, quantity=item.quantity, base=item.base, net=item.net, tax=item.tax, total=item.total, ) for item in invoice.line_items ] ) if invoice.closed and invoice.revision_of is None: BillingEntry.objects.get_or_create( entry_id=f"BE-{invoice.invoice_id}", defaults={ "account": account, "source_type": "invoice", "source_id": invoice.invoice_id, "amount": invoice.total, "period_key": invoice.period.key, "invoice_id": invoice.invoice_id, "acceptance_ordinal": None, }, ) return row def resolution(account: DomainAccount) -> None: """Write the mutated aggregate back without changing issued artifacts.""" account_row = Account.objects.get(account_id=account.account_id) for revision in account.revisions: store_invoice(account_row, revision, issued=True) for invoice in account.invoices.values(): row = Invoice.objects.get(invoice_id=invoice.invoice_id) if row.closed: continue row.total = invoice.total row.rule_set_key = invoice.rule_set.key row.selected_exclusive_promotion = invoice.selected_exclusive_promotion row.non_selected_exclusive_promotions = list( invoice.non_selected_exclusive_promotions ) row.save( update_fields=[ "total", "rule_set_key", "selected_exclusive_promotion", "non_selected_exclusive_promotions", ] ) row.line_items.all().delete() LineItem.objects.bulk_create( [ LineItem( invoice=row, usage_key=item.usage_key, description=item.description, quantity=item.quantity, base=item.base, net=item.net, tax=item.tax, total=item.total, ) for item in invoice.line_items ] ) for record in account.adjustment_records: original = Invoice.objects.get(invoice_id=record.original_invoice_id) AdjustmentRecord.objects.get_or_create( record_id=record.record_id, defaults={ "adjustment_id": record.adjustment_id, "account": account_row, "acceptance_ordinal": record.acceptance_ordinal, "original_invoice": original, "target_period": record.target_period, "delta": record.delta, "prior_recorded_total": record.prior_recorded_total, "resulting_total": record.resulting_total, "policy": record.policy, "effective_at": record.effective_at, "applied_period": record.applied_period, "replacement_invoice_id": record.replacement_invoice_id, "void_amount": record.void_amount, "reissue_amount": record.reissue_amount, "replaces_adjustment_id": record.replaces_adjustment_id, "withdraws_adjustment_id": record.withdraws_adjustment_id, }, ) for entry in account.billing_entries: BillingEntry.objects.get_or_create( entry_id=entry.entry_id, defaults={ "account": account_row, "source_type": entry.source_type, "source_id": entry.source_id, "amount": entry.amount, "period_key": entry.period_key, "invoice_id": entry.invoice_id, "acceptance_ordinal": entry.acceptance_ordinal, }, ) for run in account.statement_runs: run_row, _ = StatementRun.objects.update_or_create( run_id=run.run_id, defaults={ "account": account_row, "run_number": run.run_number, "operation_id": run.operation_id, "predecessor_run_id": run.predecessor_run_id, "previous_cutoff_ordinal": run.previous_cutoff_ordinal, "cutoff_ordinal": run.cutoff_ordinal, "status": run.status, "recorded_demand": run.recorded_demand, "incremental_demand": run.incremental_demand, }, ) for member in run.membership: record = AdjustmentRecord.objects.get(record_id=member.record_id) RunMembership.objects.get_or_create( run=run_row, record=record, defaults={ "adjustment_id": member.adjustment_id, "acceptance_ordinal": member.acceptance_ordinal, "original_invoice_id": member.original_invoice_id, }, ) if run.reconciliation_summary is not None: for item in run.reconciliation_summary: RunSummaryEntry.objects.get_or_create( run=run_row, invoice_id=item.invoice_id, defaults={ "member_count": item.member_count, "member_delta_total": item.member_delta_total, }, ) for entry in account.receivable_entries: ReceivableEntry.objects.get_or_create( run_id=entry.run_id, defaults={ "account": account_row, "predecessor_run_id": entry.predecessor_run_id, "prior_demand": entry.prior_demand, "demand": entry.demand, "incremental_demand": entry.incremental_demand, }, ) account_row.summaries_adopted_ordinal = account.summaries_adopted_ordinal account_row.summary_adoption_id = account._summary_adoption_id account_row.balance = account.balance account_row.amount_due = account.amount_due account_row.save( update_fields=[ "summaries_adopted_ordinal", "summary_adoption_id", "balance", "amount_due", ] )