From 870928e20d5ce89682c5f82a49764007e3783a8f Mon Sep 17 00:00:00 2001 From: omicron Date: Fri, 9 Jan 2026 03:59:20 +0100 Subject: [PATCH] Add routes to get weekly totals --- mft/routes/__init__.py | 3 +- mft/routes/statistics.py | 219 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 221 insertions(+), 1 deletion(-) create mode 100644 mft/routes/statistics.py diff --git a/mft/routes/__init__.py b/mft/routes/__init__.py index 518316d..c994d50 100644 --- a/mft/routes/__init__.py +++ b/mft/routes/__init__.py @@ -1,5 +1,5 @@ from fastapi import APIRouter -from mft.routes import auth, categories, expense, health +from mft.routes import auth, categories, expense, health, statistics api_router = APIRouter() @@ -7,3 +7,4 @@ api_router.include_router(health.router, tags=["health"]) api_router.include_router(auth.router, prefix="/auth", tags=["auth"]) api_router.include_router(categories.router, tags=["categories"]) api_router.include_router(expense.router, tags=["expenses"]) +api_router.include_router(statistics.router, tags=["statistics"]) diff --git a/mft/routes/statistics.py b/mft/routes/statistics.py new file mode 100644 index 0000000..5782a0e --- /dev/null +++ b/mft/routes/statistics.py @@ -0,0 +1,219 @@ +import re +from datetime import datetime, timezone, timedelta +from fastapi import APIRouter, HTTPException, status, Depends, Query +from pydantic import BaseModel +from mft.auth import verify_token +from mft.database import get_db + +router = APIRouter() + + +class CategoryTotal(BaseModel): + """Per-category total within a time period.""" + + cid: int + name: str + total: int + + +class WeekTotal(BaseModel): + """Weekly totals and per-category breakdown.""" + + week: str # ISO week format (YYYY-Www) + from_date: str # ISO date format (YYYY-MM-DD) + to_date: str # ISO date format (YYYY-MM-DD) + total: int + by_category: list[CategoryTotal] + + +def parse_iso_week(week_str: str) -> datetime: + """ + Parse ISO week format (YYYY-Www) to datetime at start of week (Monday). + + Args: + week_str: String in YYYY-Www format (e.g., "2026-W01") + + Returns: + datetime object for Monday of that week (00:00:00 UTC) + + Raises: + ValueError: If format is invalid + """ + if not re.match(r"^\d{4}-W\d{2}$", week_str): + raise ValueError(f"Invalid ISO week format: {week_str}") + + try: + # Parse ISO week: %G=ISO year, %V=ISO week, %u=day of week (1=Mon) + return datetime.strptime(f"{week_str}-1", "%G-W%V-%u").replace( + tzinfo=timezone.utc + ) + except ValueError as e: + raise ValueError(f"Invalid ISO week: {week_str}") from e + + +def get_iso_week_string(dt: datetime) -> str: + """ + Convert datetime to ISO week format string. + + Args: + dt: datetime object + + Returns: + ISO week string (YYYY-Www) + """ + iso_cal = dt.isocalendar() + return f"{iso_cal.year}-W{iso_cal.week:02d}" + + +def get_default_week_range() -> tuple[datetime, datetime]: + """ + Get default week range: The most recent 4 weeks, which includes the ongoing week. + + Returns: + Tuple of (from_date, to_date) where range is [from, to) + from_date is Monday of 3 weeks ago + to_date is Monday of next week (start of week after current) + """ + now = datetime.now(timezone.utc) + current_week_monday = now - timedelta(days=now.weekday()) + current_week_monday = current_week_monday.replace( + hour=0, minute=0, second=0, microsecond=0 + ) + + from_date = current_week_monday - timedelta(days=21) + to_date = current_week_monday + timedelta(days=7) + + return from_date, to_date + + +@router.get("/statistics/totals", response_model=list[WeekTotal]) +def get_weekly_totals( + granularity: str = Query( + "weekly", description="Time granularity (only 'weekly' supported)" + ), + from_week: str | None = Query(None, description="Start week (YYYY-Www, inclusive)"), + to_week: str | None = Query(None, description="End week (YYYY-Www, exclusive)"), + uid: int = Depends(verify_token), +): + """ + Get totals aggregated by time period. + + Currently only supports weekly granularity with ISO week format (YYYY-Www). + If from_week and to_week are omitted, defaults to the most recent 4 weeks + (including the current ongoing week). + + Args: + granularity: Time granularity (must be "weekly") + from_week: Start week in ISO format (inclusive) + to_week: End week in ISO format (exclusive) + uid: User ID from authentication token + + Returns: + List of weekly totals with global and per-category breakdowns + + Raises: + 400: Invalid parameters (unsupported granularity, format errors, from >= to) + """ + # Validate granularity + if granularity != "weekly": + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Only 'weekly' granularity is currently supported", + ) + + # Validate both-or-neither for week parameters + if (from_week is None) != (to_week is None): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Both 'from_week' and 'to_week' must be provided, or both omitted", + ) + + try: + if from_week is None: + from_date, to_date = get_default_week_range() + else: + from_date = parse_iso_week(from_week) + to_date = parse_iso_week(to_week) + + if from_date >= to_date: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="'from_week' must be before 'to_week'", + ) + except ValueError as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(e), + ) from e + + from_ts = from_date.isoformat() + to_ts = to_date.isoformat() + + with get_db() as conn: + cursor = conn.cursor() + + # FIXME: this could be better with a GROUP BY/SUM query but it works + # well enough for small datasets + cursor.execute( + """ + SELECT e.ts, e.value, e.cid, c.name + FROM expense e + JOIN category c ON e.cid = c.id + WHERE e.ts >= ? AND e.ts < ? + ORDER BY e.ts + """, + (from_ts, to_ts), + ) + + rows = cursor.fetchall() + + weeks_data = {} + for row in rows: + ts_str, value, cid, category_name = row + ts = datetime.fromisoformat(ts_str) + week_str = get_iso_week_string(ts) + + if week_str not in weeks_data: + weeks_data[week_str] = { + "total": 0, + "categories": {}, + } + + weeks_data[week_str]["total"] += value + + if cid not in weeks_data[week_str]["categories"]: + weeks_data[week_str]["categories"][cid] = { + "name": category_name, + "total": 0, + } + weeks_data[week_str]["categories"][cid]["total"] += value + + # Build response with all weeks in range (including empty ones) + result = [] + current = from_date + while current < to_date: + week_str = get_iso_week_string(current) + week_data = weeks_data.get(week_str, {"total": 0, "categories": {}}) + + week_end = current + timedelta(days=7) + + by_category = [ + CategoryTotal(cid=cid, name=cat["name"], total=cat["total"]) + for cid, cat in week_data["categories"].items() + ] + # Sort by category name for consistent output + by_category.sort(key=lambda x: x.name) + + result.append( + WeekTotal( + week=week_str, + from_date=current.date().isoformat(), + to_date=week_end.date().isoformat(), + total=week_data["total"], + by_category=by_category, + ) + ) + + current = week_end + + return result