Implement FIN-WP-0001-T06 HTTP runway read surface
Add create_runway_router() following the hub-core factory pattern, expose GET /runway/summary and /runway/monthly-burn via create_app, and add finhub serve for local operation.
This commit is contained in:
parent
0f2436b34c
commit
29dbaf1c2d
11 changed files with 270 additions and 4 deletions
3
src/fin_hub/routers/__init__.py
Normal file
3
src/fin_hub/routers/__init__.py
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
from fin_hub.routers.runway import create_runway_router
|
||||
|
||||
__all__ = ["create_runway_router"]
|
||||
100
src/fin_hub/routers/runway.py
Normal file
100
src/fin_hub/routers/runway.py
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
"""Runway read router factory (hub-core pattern)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query
|
||||
|
||||
from fin_hub.schemas.runway import MonthlyBurnRead, RunwayAlertRead, RunwayProjectionRead, RunwaySummaryRead
|
||||
from fin_hub.services.evaluate import evaluate_runway
|
||||
from fin_hub.services.ledger import default_ledger_path, get_opening_balance, monthly_summary
|
||||
LedgerPathResolver = Callable[[str | None], Path]
|
||||
|
||||
|
||||
def _default_ledger_path_resolver(ledger: str | None) -> Path:
|
||||
if ledger:
|
||||
return Path(ledger)
|
||||
env_path = os.environ.get("FIN_HUB_LEDGER")
|
||||
if env_path:
|
||||
return Path(env_path)
|
||||
return default_ledger_path()
|
||||
|
||||
|
||||
def _projection_from_dict(runway: dict) -> RunwayProjectionRead:
|
||||
months = runway["months_remaining"]
|
||||
if months == float("inf"):
|
||||
months_remaining: float | None = None
|
||||
label = "unlimited"
|
||||
else:
|
||||
months_remaining = float(months)
|
||||
label = f"{months_remaining:.1f}"
|
||||
return RunwayProjectionRead(
|
||||
current_balance=float(runway["current_balance"]),
|
||||
monthly_burn=float(runway["monthly_burn"]),
|
||||
months_remaining=months_remaining,
|
||||
months_remaining_label=label,
|
||||
alert_threshold_months=float(runway["alert_threshold_months"]),
|
||||
below_threshold=bool(runway["below_threshold"]),
|
||||
currency=str(runway["currency"]),
|
||||
computed_at=str(runway["computed_at"]),
|
||||
)
|
||||
|
||||
|
||||
def create_runway_router(
|
||||
*,
|
||||
prefix: str = "/runway",
|
||||
ledger_path_resolver: LedgerPathResolver | None = None,
|
||||
) -> APIRouter:
|
||||
resolve_ledger = ledger_path_resolver or _default_ledger_path_resolver
|
||||
router = APIRouter(prefix=prefix, tags=["runway"])
|
||||
|
||||
@router.get("/monthly-burn", response_model=list[MonthlyBurnRead])
|
||||
def get_monthly_burn(
|
||||
ledger: str | None = Query(default=None, description="Ledger database path override"),
|
||||
currency: str | None = Query(default=None, description="Filter rollups by currency"),
|
||||
) -> list[MonthlyBurnRead]:
|
||||
ledger_path = resolve_ledger(ledger)
|
||||
if not ledger_path.exists():
|
||||
raise HTTPException(status_code=404, detail=f"Ledger not found: {ledger_path}")
|
||||
rollups = monthly_summary(ledger_path=ledger_path)
|
||||
if currency:
|
||||
rollups = [rollup for rollup in rollups if rollup.currency == currency.upper()]
|
||||
return [MonthlyBurnRead(**rollup.as_dict()) for rollup in rollups]
|
||||
|
||||
@router.get("/summary", response_model=RunwaySummaryRead)
|
||||
def get_runway_summary(
|
||||
ledger: str | None = Query(default=None, description="Ledger database path override"),
|
||||
balance: float | None = Query(default=None, description="Override stored opening balance"),
|
||||
threshold: float = Query(default=3.0, ge=0.0),
|
||||
currency: str = Query(default="EUR", min_length=3, max_length=3),
|
||||
allocated: float | None = Query(default=None, ge=0.0),
|
||||
spent: float | None = Query(default=None, ge=0.0),
|
||||
) -> RunwaySummaryRead:
|
||||
ledger_path = resolve_ledger(ledger)
|
||||
if not ledger_path.exists():
|
||||
raise HTTPException(status_code=404, detail=f"Ledger not found: {ledger_path}")
|
||||
try:
|
||||
report = evaluate_runway(
|
||||
ledger_path=ledger_path,
|
||||
opening_balance=balance,
|
||||
alert_threshold_months=threshold,
|
||||
currency=currency.upper(),
|
||||
allocated=allocated,
|
||||
spent=spent,
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
||||
|
||||
stored_balance, stored_currency = get_opening_balance(ledger_path)
|
||||
return RunwaySummaryRead(
|
||||
runway=_projection_from_dict(report["runway"]),
|
||||
alerts=[RunwayAlertRead(**alert) for alert in report["alerts"]],
|
||||
opening_balance=balance if balance is not None else stored_balance,
|
||||
opening_balance_currency=currency.upper() if balance is not None else stored_currency,
|
||||
ledger_path=str(ledger_path.resolve()),
|
||||
)
|
||||
|
||||
return router
|
||||
Loading…
Add table
Add a link
Reference in a new issue