Coverage for app/expenses/routes.py: 100%
136 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-06-28 23:33 +0000
1from datetime import date as _date, timedelta
3from flask import ( # pyright: ignore[reportMissingImports]
4 Blueprint,
5 abort,
6 flash,
7 redirect,
8 render_template,
9 request,
10 session,
11 url_for,
12)
13from flask.typing import ResponseReturnValue # pyright: ignore[reportMissingImports]
15from flask_babel import gettext as _ # pyright: ignore[reportMissingImports]
17from typing import Any
19from models import Aircraft, Expense, ExpenseType, FlightEntry, Role, TenantUser, db # pyright: ignore[reportMissingImports]
20from utils import login_required, require_role, user_can_access_aircraft # pyright: ignore[reportMissingImports]
22expenses_bp = Blueprint("expenses", __name__)
24_OWNER_ROLES = (Role.ADMIN, Role.OWNER)
26_CURRENCIES = ["EUR", "USD", "GBP", "CHF"]
27_UNITS = ["L", "gal"]
28_DEFAULT_PERIOD = 12 # months
31def _tenant_id() -> int:
32 tu = TenantUser.query.filter_by(user_id=session["user_id"]).first()
33 if not tu:
34 abort(403)
35 return int(tu.tenant_id)
38def _get_aircraft_or_404(aircraft_id: int) -> Aircraft:
39 ac = db.session.get(Aircraft, aircraft_id)
40 if (
41 not ac
42 or ac.tenant_id != _tenant_id()
43 or not user_can_access_aircraft(aircraft_id)
44 ):
45 abort(404)
46 return ac
49def _get_expense_or_404(aircraft: Aircraft, expense_id: int) -> Expense:
50 exp = db.session.get(Expense, expense_id)
51 if not exp or exp.aircraft_id != aircraft.id:
52 abort(404)
53 return exp
56def _compute_stats(
57 expenses: list[Any], aircraft_id: int, period_months: int
58) -> tuple[float, float | None, str]:
59 """Return (total_cost, cost_per_hour, period_label) for the filtered expense list."""
60 total_cost = sum(float(e.amount) for e in expenses)
62 if period_months > 0:
63 cutoff = _date.today() - timedelta(days=period_months * 30)
64 flights = FlightEntry.query.filter(
65 FlightEntry.aircraft_id == aircraft_id,
66 FlightEntry.date >= cutoff,
67 ).all()
68 period_label = f"last {period_months} months"
69 else:
70 flights = FlightEntry.query.filter_by(aircraft_id=aircraft_id).all()
71 period_label = "all time"
73 total_hours = sum(
74 float(f.flight_time_counter_end) - float(f.flight_time_counter_start)
75 for f in flights
76 if f.flight_time_counter_end is not None
77 and f.flight_time_counter_start is not None
78 )
79 cost_per_hour = round(total_cost / total_hours, 2) if total_hours > 0 else None
80 return total_cost, cost_per_hour, period_label
83# ── Expense list ──────────────────────────────────────────────────────────────
86@expenses_bp.route("/aircraft/<int:aircraft_id>/expenses")
87@login_required
88def list_expenses(aircraft_id: int) -> ResponseReturnValue:
89 ac = _get_aircraft_or_404(aircraft_id)
91 type_filter = request.args.get("type", "")
92 try:
93 period_months = int(request.args.get("period", _DEFAULT_PERIOD))
94 except ValueError:
95 period_months = _DEFAULT_PERIOD
97 query = Expense.query.filter_by(aircraft_id=ac.id)
99 if type_filter and type_filter in ExpenseType.ALL:
100 query = query.filter_by(expense_type=type_filter)
102 if period_months > 0:
103 cutoff = _date.today() - timedelta(days=period_months * 30)
104 query = query.filter(Expense.date >= cutoff)
106 expenses = query.order_by(Expense.date.desc(), Expense.id.desc()).all()
107 total_cost, cost_per_hour, period_label = _compute_stats(
108 expenses, ac.id, period_months
109 )
111 return render_template(
112 "expenses/list.html",
113 aircraft=ac,
114 expenses=expenses,
115 type_filter=type_filter,
116 period_months=period_months,
117 total_cost=total_cost,
118 cost_per_hour=cost_per_hour,
119 period_label=period_label,
120 expense_type_labels=ExpenseType.LABELS,
121 currencies=_CURRENCIES,
122 )
125# ── Add expense ───────────────────────────────────────────────────────────────
128@expenses_bp.route("/aircraft/<int:aircraft_id>/expenses/add", methods=["GET", "POST"])
129@login_required
130@require_role(*_OWNER_ROLES)
131def add_expense(aircraft_id: int) -> ResponseReturnValue:
132 ac = _get_aircraft_or_404(aircraft_id)
134 if request.method == "POST":
135 err = _validate_and_save(ac, expense=None)
136 if err is None:
137 flash(_("Expense recorded."), "success")
138 return redirect(url_for("expenses.list_expenses", aircraft_id=ac.id))
139 flash(err, "danger")
141 return render_template(
142 "expenses/expense_form.html",
143 aircraft=ac,
144 expense=None,
145 expense_types=ExpenseType.LABELS,
146 currencies=_CURRENCIES,
147 units=_UNITS,
148 today=_date.today().isoformat(),
149 )
152# ── Edit expense ──────────────────────────────────────────────────────────────
155@expenses_bp.route(
156 "/aircraft/<int:aircraft_id>/expenses/<int:expense_id>/edit",
157 methods=["GET", "POST"],
158)
159@login_required
160@require_role(*_OWNER_ROLES)
161def edit_expense(aircraft_id: int, expense_id: int) -> ResponseReturnValue:
162 ac = _get_aircraft_or_404(aircraft_id)
163 exp = _get_expense_or_404(ac, expense_id)
165 if request.method == "POST":
166 err = _validate_and_save(ac, expense=exp)
167 if err is None:
168 flash(_("Expense updated."), "success")
169 return redirect(url_for("expenses.list_expenses", aircraft_id=ac.id))
170 flash(err, "danger")
172 return render_template(
173 "expenses/expense_form.html",
174 aircraft=ac,
175 expense=exp,
176 expense_types=ExpenseType.LABELS,
177 currencies=_CURRENCIES,
178 units=_UNITS,
179 today=_date.today().isoformat(),
180 )
183# ── Delete expense ────────────────────────────────────────────────────────────
186@expenses_bp.route(
187 "/aircraft/<int:aircraft_id>/expenses/<int:expense_id>/delete", methods=["POST"]
188)
189@login_required
190@require_role(*_OWNER_ROLES)
191def delete_expense(aircraft_id: int, expense_id: int) -> ResponseReturnValue:
192 ac = _get_aircraft_or_404(aircraft_id)
193 exp = _get_expense_or_404(ac, expense_id)
194 db.session.delete(exp)
195 db.session.commit()
196 flash(_("Expense deleted."), "success")
197 return redirect(url_for("expenses.list_expenses", aircraft_id=ac.id))
200# ── Shared save helper ────────────────────────────────────────────────────────
203def _validate_and_save(aircraft: Aircraft, expense: Expense | None) -> str | None:
204 """Validate POST data, persist, return error string or None on success."""
205 date_str = request.form.get("date", "").strip()
206 expense_type = request.form.get("expense_type", "").strip()
207 description = request.form.get("description", "").strip() or None
208 amount_str = request.form.get("amount", "").strip()
209 currency = request.form.get("currency", "EUR").strip()
210 quantity_str = request.form.get("quantity", "").strip()
211 unit = request.form.get("unit", "").strip() or None
213 if not date_str:
214 return str(_("Date is required."))
215 try:
216 from datetime import date as _date_cls
218 date_val = _date_cls.fromisoformat(date_str)
219 except ValueError:
220 return str(_("Invalid date format."))
222 if expense_type not in ExpenseType.ALL:
223 return str(_("Invalid expense type."))
225 if not amount_str:
226 return str(_("Amount is required."))
227 try:
228 amount = float(amount_str)
229 if amount < 0:
230 raise ValueError
231 except ValueError:
232 return str(_("Amount must be a non-negative number."))
234 quantity = None
235 if quantity_str:
236 try:
237 quantity = float(quantity_str)
238 if quantity < 0:
239 raise ValueError
240 except ValueError:
241 return str(_("Quantity must be a non-negative number."))
243 if expense is None:
244 expense = Expense(aircraft_id=aircraft.id)
245 db.session.add(expense)
247 expense.date = date_val
248 expense.expense_type = expense_type
249 expense.description = description
250 expense.amount = amount
251 expense.currency = currency
252 expense.quantity = quantity
253 expense.unit = unit if quantity else None
254 db.session.commit()
255 return None