Add SQLite passive income dashboard with Excel import and deployment tools

app/routes/income.py

@@ -0,0 +1,91 @@
+from datetime import date
+from decimal import Decimal
+from typing import Annotated
+from fastapi import APIRouter, Form, HTTPException, Query, Request
+from fastapi.responses import RedirectResponse
+from database import connect, ensure_asset
+from models import CATEGORIES
+from services.income_service import assets, available_years, get_entry, list_entries, save_entry
+from views import render
+
+router = APIRouter()
+
+
+def form_page(request, data, error=None, status=200, entry_id=None):
+    return render(request, 'income_form.html', {'data': data, 'assets': assets(), 'error': error, 'entry_id': entry_id}, status)
+
+
+@router.get('/income')
+def history(request: Request, year: str | None = None,
+            month: str | None = None,
+            asset_id: str | None = None, category: str | None = None,
+            page: Annotated[int, Query(ge=1, le=1000000)] = 1):
+    def number(value, maximum):
+        if not value:
+            return None
+        try:
+            parsed = int(value)
+            if not 1 <= parsed <= maximum:
+                raise ValueError
+            return parsed
+        except ValueError:
+            raise HTTPException(422, 'Ungültiger Filter.') from None
+    year, month, asset_id = number(year, 9999), number(month, 12), number(asset_id, 9223372036854775807)
+    category = category or None
+    if category is not None and category not in CATEGORIES:
+        raise HTTPException(422, 'Ungültige Kategorie.')
+    rows = list_entries(year, month, asset_id, category, limit=101, offset=(page-1)*100)
+    return render(request, 'income.html', dict(entries=rows[:100], more=len(rows)>100, page=page,
+                  assets=assets(), years=available_years(), filters=dict(year=year, month=month, asset_id=asset_id, category=category)))
+
+
+@router.get('/income/new')
+def new(request: Request):
+    return form_page(request, {'date': date.today().isoformat(), 'received': True, 'expected': False,
+                               'asset_id': request.query_params.get('asset_id', ''), 'category': 'dividend'})
+
+
+@router.get('/income/{entry_id}/edit')
+def edit(request: Request, entry_id: int):
+    data = get_entry(entry_id)
+    data['amount'] = str(Decimal(data['amount']) / 100).replace('.', ',')
+    return form_page(request, data, entry_id=entry_id)
+
+
+@router.post('/income/new')
+@router.post('/income/{entry_id}/edit')
+def save(request: Request, date: Annotated[str, Form()] = '', asset_id: Annotated[str, Form()] = '',
+         category: Annotated[str, Form()] = '', amount: Annotated[str, Form()] = '',
+         note: Annotated[str, Form()] = '', expected: Annotated[str, Form()] = '',
+         received: Annotated[str, Form()] = '', entry_id: int | None = None):
+    data = dict(date=date, asset_id=asset_id, category=category, amount=amount, note=note, expected=expected, received=received)
+    try:
+        save_entry(data, entry_id)
+    except ValueError as error:
+        return form_page(request, data, str(error), 422, entry_id)
+    return RedirectResponse('/?message=saved', status_code=303)
+
+
+@router.post('/income/{entry_id}/delete')
+def delete(entry_id: int):
+    if not 1 <= entry_id <= 9223372036854775807:
+        raise HTTPException(404, 'Zahlung nicht gefunden.')
+    with connect() as db:
+        if db.execute('DELETE FROM income_entries WHERE id = ?', (entry_id,)).rowcount == 0:
+            raise HTTPException(404, 'Zahlung nicht gefunden.')
+    return RedirectResponse('/?message=deleted', status_code=303)
+
+
+@router.get('/assets/new')
+def new_asset(request: Request):
+    return render(request, 'asset_form.html', {'data': {}})
+
+
+@router.post('/assets/new')
+def create_asset(request: Request, name: Annotated[str, Form()], asset_type: Annotated[str, Form()], ticker: Annotated[str, Form()] = ''):
+    try:
+        with connect() as db:
+            asset_id = ensure_asset(db, name, asset_type, ticker)
+    except ValueError as error:
+        return render(request, 'asset_form.html', {'data': dict(name=name, asset_type=asset_type, ticker=ticker), 'error': str(error)}, 422)
+    return RedirectResponse(f'/income/new?asset_id={asset_id}', status_code=303)