Coverage for app/services/billing.py: 100%
104 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-31 10:13 +0000
1"""Shared billing core — Phases 37 (rental), 38 (shared ownership), 39 (flying
2club). See docs/billing_service_design.md for the full design rationale.
4An account per person/scope, an append-only ledger of charges and payments,
5a derived balance, and a period statement. All writes go through this
6service; routes never insert LedgerEntry rows directly. Callers own the
7transaction — nothing here commits.
8"""
10from __future__ import annotations
12import csv
13import io
14from dataclasses import dataclass, field
15from datetime import UTC, date, timedelta
16from decimal import ROUND_HALF_UP, Decimal
17from typing import TYPE_CHECKING, Any, cast
19if TYPE_CHECKING:
20 from models import BillingAccount, LedgerEntry, User
22TWO_PLACES = Decimal("0.01")
25def _quantize(amount: Any) -> Decimal:
26 return Decimal(str(amount)).quantize(TWO_PLACES, rounding=ROUND_HALF_UP)
29@dataclass
30class StatementLine:
31 entry: LedgerEntry
32 running_balance: Decimal
35@dataclass
36class Statement:
37 account: BillingAccount
38 start: date
39 end: date
40 opening_balance: Decimal
41 closing_balance: Decimal
42 lines: list[StatementLine] = field(default_factory=list)
45class BillingService:
46 @staticmethod
47 def get_or_create_account(
48 tenant_id: int, user_id: int, kind: str, aircraft_id: int | None = None
49 ) -> BillingAccount:
50 """Return the existing account for this (tenant, user, kind, aircraft)
51 scope, or create it. Idempotent under a concurrent-request race —
52 the unique constraint is the source of truth, not a pre-check."""
53 from models import BillingAccount, db
54 from sqlalchemy.exc import IntegrityError
56 existing: BillingAccount | None = BillingAccount.query.filter_by(
57 tenant_id=tenant_id, user_id=user_id, kind=kind, aircraft_id=aircraft_id
58 ).first()
59 if existing is not None:
60 return existing
62 account = BillingAccount(
63 tenant_id=tenant_id, user_id=user_id, kind=kind, aircraft_id=aircraft_id
64 )
65 db.session.add(account)
66 try:
67 db.session.flush()
68 except IntegrityError:
69 db.session.rollback()
70 existing = BillingAccount.query.filter_by(
71 tenant_id=tenant_id,
72 user_id=user_id,
73 kind=kind,
74 aircraft_id=aircraft_id,
75 ).first()
76 if existing is None: # pragma: no cover — defensive, race lost twice
77 raise
78 return existing
79 return account
81 @staticmethod
82 def _insert(
83 account: BillingAccount,
84 entry_type: str,
85 amount: Any,
86 description: str,
87 entry_date: date,
88 source_type: str | None = None,
89 source_id: int | None = None,
90 created_by: User | None = None,
91 reverses_id: int | None = None,
92 ) -> LedgerEntry:
93 from models import LedgerEntry, db
95 entry = LedgerEntry(
96 account_id=account.id,
97 entry_type=entry_type,
98 amount=_quantize(amount),
99 description=description,
100 entry_date=entry_date,
101 source_type=source_type,
102 source_id=source_id,
103 reverses_id=reverses_id,
104 created_by_id=created_by.id if created_by is not None else None,
105 )
106 db.session.add(entry)
107 return entry
109 @staticmethod
110 def post(
111 account: BillingAccount,
112 entry_type: str,
113 amount: Any,
114 description: str,
115 entry_date: date,
116 source_type: str | None = None,
117 source_id: int | None = None,
118 created_by: User | None = None,
119 ) -> LedgerEntry:
120 """Validates sign against entry_type; commits nothing (caller owns
121 the transaction)."""
122 from models import LedgerEntryType
124 if entry_type not in LedgerEntryType.ALL:
125 raise ValueError(f"Unknown entry_type: {entry_type!r}")
127 quantized = _quantize(amount)
128 if entry_type == LedgerEntryType.CHARGE and quantized <= 0:
129 raise ValueError("CHARGE amount must be > 0")
130 if (
131 entry_type in (LedgerEntryType.PAYMENT, LedgerEntryType.CREDIT)
132 and quantized >= 0
133 ):
134 raise ValueError(f"{entry_type} amount must be < 0")
135 if entry_type == LedgerEntryType.ADJUSTMENT and not description:
136 raise ValueError("ADJUSTMENT requires a non-empty description")
138 return BillingService._insert(
139 account,
140 entry_type,
141 quantized,
142 description,
143 entry_date,
144 source_type=source_type,
145 source_id=source_id,
146 created_by=created_by,
147 )
149 @staticmethod
150 def reverse(entry: LedgerEntry, created_by: User | None, note: str) -> LedgerEntry:
151 """Posts the mirror entry (same type, opposite amount) with
152 reverses_id set. Refuses to reverse a reversal and refuses to
153 reverse the same entry twice."""
154 from models import LedgerEntry
156 if entry.reverses_id is not None:
157 raise ValueError("Cannot reverse a reversal entry")
158 already_reversed = LedgerEntry.query.filter_by(reverses_id=entry.id).first()
159 if already_reversed is not None:
160 raise ValueError("This entry has already been reversed")
162 return BillingService._insert(
163 cast("BillingAccount", entry.account),
164 entry.entry_type,
165 -Decimal(entry.amount),
166 note,
167 entry.entry_date,
168 source_type=entry.source_type,
169 source_id=entry.source_id,
170 created_by=created_by,
171 reverses_id=entry.id,
172 )
174 @staticmethod
175 def balance(account: BillingAccount, as_of: date | None = None) -> Decimal:
176 from models import LedgerEntry, db
178 query = db.session.query(db.func.sum(LedgerEntry.amount)).filter(
179 LedgerEntry.account_id == account.id
180 )
181 if as_of is not None:
182 query = query.filter(LedgerEntry.entry_date <= as_of)
183 total = query.scalar()
184 return _quantize(total) if total is not None else Decimal("0.00")
186 @staticmethod
187 def statement(account: BillingAccount, start: date, end: date) -> Statement:
188 """Opening balance (sum of entries before start), chronological
189 entries in [start, end], closing balance."""
190 from models import LedgerEntry
192 opening = BillingService.balance(account, as_of=start - timedelta(days=1))
193 entries = (
194 LedgerEntry.query.filter(
195 LedgerEntry.account_id == account.id,
196 LedgerEntry.entry_date >= start,
197 LedgerEntry.entry_date <= end,
198 )
199 .order_by(LedgerEntry.entry_date, LedgerEntry.id)
200 .all()
201 )
202 running = opening
203 lines = []
204 for entry in entries:
205 running = _quantize(running + Decimal(entry.amount))
206 lines.append(StatementLine(entry=entry, running_balance=running))
207 return Statement(
208 account=account,
209 start=start,
210 end=end,
211 opening_balance=opening,
212 closing_balance=running,
213 lines=lines,
214 )
216 @staticmethod
217 def statement_csv(statement: Statement, exported_by: User | None = None) -> str:
218 """Header rows: export date, exporter, period, account holder, scope.
219 Then one row per entry: date, type, description, amount, running
220 balance."""
221 from datetime import datetime
223 buf = io.StringIO()
224 writer = csv.writer(buf)
225 writer.writerow(["Export date", datetime.now(UTC).date().isoformat()])
226 writer.writerow(["Exporter", exported_by.display_name if exported_by else ""])
227 writer.writerow(
228 ["Period", f"{statement.start.isoformat()} to {statement.end.isoformat()}"]
229 )
230 account = statement.account
231 writer.writerow(
232 ["Account holder", account.user.display_name if account.user else ""]
233 )
234 writer.writerow(["Scope", account.kind])
235 writer.writerow([])
236 writer.writerow(["Opening balance", "", "", "", str(statement.opening_balance)])
237 writer.writerow(["Date", "Type", "Description", "Amount", "Running balance"])
238 for line in statement.lines:
239 writer.writerow(
240 [
241 line.entry.entry_date.isoformat(),
242 line.entry.entry_type,
243 line.entry.description,
244 str(line.entry.amount),
245 str(line.running_balance),
246 ]
247 )
248 writer.writerow(["Closing balance", "", "", "", str(statement.closing_balance)])
249 return buf.getvalue()