Jakeandopencode
aced0a1a3b
- Renewal tracking: compute due/overdue status from the most recent mot/insurance event (renewal_months, renewal_due_soon_days settings), surfaced on the service card and dashboard. - Service-status card metrics now render as a two-column label/value table with row separators instead of wrapping auto-fill grid cells. - Add a bicycle favicon (favicon.svg) to base + login templates. Co-Authored-By: opencode <opencode@users.noreply.github.com>
511 lines
16 KiB
Python
511 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import asynccontextmanager
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, Request, Response, status
|
|
from fastapi.templating import Jinja2Templates
|
|
from pydantic import BaseModel, field_validator
|
|
from starlette.responses import RedirectResponse
|
|
from starlette.staticfiles import StaticFiles
|
|
|
|
from . import db as d
|
|
from . import services as svc
|
|
from .config import Settings, get_settings
|
|
from .security import (
|
|
SESSION_COOKIE,
|
|
check_csrf,
|
|
client_ip,
|
|
create_session,
|
|
csrf_token,
|
|
delete_session,
|
|
get_user_id_for_session,
|
|
hash_password,
|
|
is_throttled,
|
|
record_login_attempt,
|
|
verify_password,
|
|
)
|
|
|
|
APP_DIR = Path(__file__).resolve().parent
|
|
templates = Jinja2Templates(directory=str(APP_DIR / "templates"))
|
|
|
|
|
|
# --- shared helpers --------------------------------------------
|
|
def _s(request: Request) -> Settings:
|
|
return get_settings()
|
|
|
|
|
|
def _redirect(location: str, status_code: int = status.HTTP_303_SEE_OTHER):
|
|
return RedirectResponse(location, status_code=status_code)
|
|
|
|
|
|
def _template_env(request: Request) -> dict:
|
|
s = get_settings()
|
|
base: dict = {
|
|
"request": request,
|
|
"event_types": s.event_types,
|
|
"today": date.today(),
|
|
"settings": s,
|
|
}
|
|
return base
|
|
|
|
|
|
def _csrf(request: Request) -> str:
|
|
s = get_settings()
|
|
return csrf_token(s.secret_key, request.cookies.get(SESSION_COOKIE))
|
|
|
|
|
|
def _parse_int(value: str | None) -> int | None:
|
|
if value is None:
|
|
return None
|
|
value = value.strip().replace(",", "")
|
|
if not value:
|
|
return None
|
|
return int(value)
|
|
|
|
|
|
def _errors_from(exc: Exception) -> dict[str, str]:
|
|
if hasattr(exc, "errors"):
|
|
return {e["loc"][-1]: e["msg"] for e in exc.errors()}
|
|
return {"_": str(exc)}
|
|
|
|
|
|
def current_user_id(request: Request) -> int | None:
|
|
token = request.cookies.get(SESSION_COOKIE)
|
|
with d.get_conn() as conn:
|
|
return get_user_id_for_session(conn, token)
|
|
|
|
|
|
def require_user(request: Request) -> int:
|
|
uid = current_user_id(request)
|
|
if uid is None:
|
|
raise FastAPIUnauthorized()
|
|
return uid
|
|
|
|
|
|
class FastAPIUnauthorized(Exception):
|
|
pass
|
|
|
|
|
|
# --- form models -------------------------------------------------
|
|
class BikeIn(BaseModel):
|
|
name: str
|
|
service_interval_miles: int | None = None
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def _name(cls, v: str) -> str:
|
|
v = v.strip()
|
|
if not v or len(v) > 100:
|
|
raise ValueError("Name must be 1-100 characters.")
|
|
return v
|
|
|
|
@field_validator("service_interval_miles")
|
|
@classmethod
|
|
def _interval(cls, v: int | None) -> int | None:
|
|
if v is not None and not (1 <= v <= 10_000_000):
|
|
raise ValueError("Interval must be a positive number of miles, or blank.")
|
|
return v
|
|
|
|
|
|
class EventIn(BaseModel):
|
|
type: str
|
|
event_date: date
|
|
mileage: int | None = None
|
|
cost: int | None = None
|
|
notes: str | None = None
|
|
|
|
@field_validator("type")
|
|
@classmethod
|
|
def _type(cls, v: str) -> str:
|
|
s = get_settings()
|
|
if v not in s.event_types:
|
|
raise ValueError(f"Type must be one of: {', '.join(s.event_types)}.")
|
|
return v
|
|
|
|
@field_validator("event_date")
|
|
@classmethod
|
|
def _date(cls, v: date) -> date:
|
|
if v > date.today():
|
|
raise ValueError("Date cannot be in the future.")
|
|
if v < date(1950, 1, 1):
|
|
raise ValueError("Date looks too old.")
|
|
return v
|
|
|
|
@field_validator("mileage")
|
|
@classmethod
|
|
def _mileage(cls, v: int | None) -> int | None:
|
|
if v is not None and v < 0:
|
|
raise ValueError("Mileage cannot be negative.")
|
|
return v
|
|
|
|
@field_validator("cost")
|
|
@classmethod
|
|
def _cost(cls, v: int | None) -> int | None:
|
|
if v is not None and v < 0:
|
|
raise ValueError("Cost cannot be negative.")
|
|
return v
|
|
|
|
|
|
def _form_str(form, key: str) -> str:
|
|
return str(form.get(key) or "").strip()
|
|
|
|
|
|
def _form_int(form, key: str) -> int | None:
|
|
return _parse_int(_form_str(form, key))
|
|
|
|
|
|
def _form_date(form, key: str) -> date:
|
|
return date.fromisoformat(_form_str(form, key) or date.today().isoformat())
|
|
|
|
|
|
def _form_notes(form) -> str | None:
|
|
v = _form_str(form, "notes")
|
|
return v or None
|
|
|
|
|
|
def _bike_in(form) -> BikeIn:
|
|
return BikeIn(
|
|
name=_form_str(form, "name"),
|
|
service_interval_miles=_form_int(form, "service_interval_miles"),
|
|
)
|
|
|
|
|
|
def _event_in(form) -> EventIn:
|
|
return EventIn(
|
|
type=_form_str(form, "type"),
|
|
event_date=_form_date(form, "event_date"),
|
|
mileage=_form_int(form, "mileage"),
|
|
cost=_form_int(form, "cost"),
|
|
notes=_form_notes(form),
|
|
)
|
|
|
|
|
|
# --- request context helpers (form data -> dict for re-render) ---
|
|
def _form_dict(form) -> dict:
|
|
"""Best-effort form dict for pre-fill on validation errors."""
|
|
out: dict = {}
|
|
try:
|
|
for key, value in form.items():
|
|
out[key] = value if isinstance(value, str) else str(value)
|
|
except AttributeError:
|
|
pass
|
|
return out
|
|
|
|
|
|
# --- app factory -------------------------------------------------
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
s = get_settings()
|
|
s.data_dir.mkdir(parents=True, exist_ok=True)
|
|
d.init_schema(s.resolved_db_path)
|
|
if s.password is None:
|
|
raise RuntimeError(
|
|
"BIKE_ADMIN_PASSWORD is not set. Refusing to start to avoid an "
|
|
"insecure default. Set BIKE_ADMIN_USERNAME and BIKE_ADMIN_PASSWORD "
|
|
"in the environment (or a .env file) and restart."
|
|
)
|
|
d.ensure_user(
|
|
s.resolved_db_path,
|
|
s.username,
|
|
s.password,
|
|
hash_password(s.password),
|
|
)
|
|
yield
|
|
|
|
|
|
app = FastAPI(title="bike-admin", lifespan=lifespan)
|
|
|
|
|
|
@app.exception_handler(FastAPIUnauthorized)
|
|
async def _unauthorized(request: Request, exc: FastAPIUnauthorized):
|
|
return RedirectResponse("/login", status_code=303)
|
|
|
|
|
|
app.mount("/static", StaticFiles(directory=str(APP_DIR / "static")), name="static")
|
|
|
|
templates.env.filters["money"] = lambda v: f"{v:,}" if v is not None else ""
|
|
|
|
templates.env.globals["event_types"] = get_settings().event_types
|
|
templates.env.globals["status_class"] = svc.status_class
|
|
templates.env.globals["status_text"] = svc.status_text
|
|
templates.env.globals["renewal_class"] = svc.renewal_class
|
|
templates.env.globals["renewal_text"] = svc.renewal_text
|
|
|
|
|
|
@app.middleware("http")
|
|
async def security_headers(request: Request, call_next):
|
|
response = await call_next(request)
|
|
h = response.headers
|
|
h.setdefault("X-Content-Type-Options", "nosniff")
|
|
h.setdefault("X-Frame-Options", "DENY")
|
|
h.setdefault("Referrer-Policy", "same-origin")
|
|
h.setdefault("Permissions-Policy", "geolocation=(), camera=(), microphone=()")
|
|
csp = (
|
|
"default-src 'self'; "
|
|
"img-src 'self' data:; "
|
|
"style-src 'self'; "
|
|
"font-src 'self'; "
|
|
"script-src 'self'; "
|
|
"form-action 'self'; "
|
|
"base-uri 'self'; "
|
|
"frame-ancestors 'none'"
|
|
)
|
|
h.setdefault("Content-Security-Policy", csp)
|
|
return response
|
|
|
|
|
|
def _cookie_opts(request: Request, s: Settings) -> dict:
|
|
secure = bool(s.is_production) or (request.url is not None and request.url.scheme == "https")
|
|
return {
|
|
"path": "/",
|
|
"samesite": "lax",
|
|
"httponly": True,
|
|
"secure": secure,
|
|
"max_age": s.session_ttl_seconds,
|
|
}
|
|
|
|
|
|
def _csrf_ok(request: Request, form) -> bool:
|
|
s = get_settings()
|
|
token = request.cookies.get(SESSION_COOKIE)
|
|
return check_csrf(s.secret_key, token, _form_str(form, "csrf"))
|
|
|
|
|
|
def _login_render(request: Request, error: str | None) -> Response:
|
|
context = _template_env(request)
|
|
return templates.TemplateResponse(request, "login.html", {**context, "error": error})
|
|
|
|
|
|
def _bike_render(
|
|
request: Request, kind: str, bike, errors: dict, old: dict, action: str | None = None
|
|
) -> Response:
|
|
is_edit = kind == "edit"
|
|
if action is None:
|
|
action = request.url.path
|
|
context = _template_env(request)
|
|
context.update(
|
|
bike=bike,
|
|
errors=errors,
|
|
old=old,
|
|
csrf=_csrf(request),
|
|
action=action,
|
|
is_edit=is_edit,
|
|
)
|
|
return templates.TemplateResponse(request, f"bikes/{kind}.html", context)
|
|
|
|
|
|
# --- auth routes -------------------------------------------------
|
|
@app.get("/login", include_in_schema=False)
|
|
async def login_get(request: Request):
|
|
return _login_render(request, None)
|
|
|
|
|
|
@app.post("/login", include_in_schema=False)
|
|
async def login_post(request: Request):
|
|
s = get_settings()
|
|
form = await request.form()
|
|
username = _form_str(form, "username")
|
|
password = _form_str(form, "password")
|
|
ip = client_ip(request, s.trust_proxy)
|
|
|
|
with d.get_conn() as conn:
|
|
if is_throttled(conn, ip, s):
|
|
return _login_render(request, "Too many login attempts. Try again in a few minutes.")
|
|
cur = conn.execute(
|
|
"SELECT id, password_hash FROM users WHERE username = ?", (username,)
|
|
).fetchone()
|
|
ok = cur is not None and verify_password(password, cur["password_hash"])
|
|
if not ok:
|
|
record_login_attempt(conn, ip)
|
|
return _login_render(request, "Invalid username or password.")
|
|
token = create_session(conn, cur["id"], s)
|
|
|
|
nxt = _form_str(form, "next") or "/"
|
|
resp = RedirectResponse(nxt, status_code=303)
|
|
resp.set_cookie(SESSION_COOKIE, token, **_cookie_opts(request, s))
|
|
return resp
|
|
|
|
|
|
@app.get("/logout", include_in_schema=False)
|
|
async def logout_get(request: Request):
|
|
token = request.cookies.get(SESSION_COOKIE)
|
|
with d.get_conn() as conn:
|
|
delete_session(conn, token)
|
|
resp = _login_render(request, None)
|
|
resp.delete_cookie(SESSION_COOKIE, path="/")
|
|
return resp
|
|
|
|
|
|
# --- dashboard ---------------------------------------------------
|
|
@app.get("/", include_in_schema=False)
|
|
async def dashboard(request: Request):
|
|
require_user(request)
|
|
with d.get_conn() as conn:
|
|
action = svc.action_needed(conn)
|
|
soon = svc.due_soon(conn)
|
|
bikes = svc.all_summaries(conn)
|
|
context = _template_env(request)
|
|
context.update(
|
|
action=action,
|
|
due_soon=soon,
|
|
bikes=bikes,
|
|
csrf=_csrf(request),
|
|
)
|
|
return templates.TemplateResponse(request, "dashboard.html", context)
|
|
|
|
|
|
# --- bikes -------------------------------------------------------
|
|
@app.get("/bikes/new", include_in_schema=False)
|
|
async def bike_new_get(request: Request):
|
|
require_user(request)
|
|
return _bike_render(request, "new", None, {}, {})
|
|
|
|
|
|
@app.post("/bikes/new", include_in_schema=False)
|
|
async def bike_new_post(request: Request):
|
|
require_user(request)
|
|
form = await request.form()
|
|
if not _csrf_ok(request, form):
|
|
return _bike_render(request, "new", None, {"_": "CSRF check failed."}, _form_dict(form))
|
|
|
|
try:
|
|
bike_in = _bike_in(form)
|
|
except Exception as exc:
|
|
return _bike_render(request, "new", None, _errors_from(exc), _form_dict(form))
|
|
|
|
with d.get_conn() as conn:
|
|
bike_id = d.create_bike(conn, bike_in.name, bike_in.service_interval_miles)
|
|
return _redirect(f"/bikes/{bike_id}")
|
|
|
|
|
|
@app.get("/bikes/{bike_id}", include_in_schema=False)
|
|
async def bike_detail_get(request: Request, bike_id: int):
|
|
require_user(request)
|
|
s = get_settings()
|
|
with d.get_conn() as conn:
|
|
bike = d.get_bike(conn, bike_id)
|
|
if bike is None:
|
|
return _redirect("/")
|
|
summary = svc.summarize_bike(conn, bike)
|
|
events = d.list_events(conn, bike.id)
|
|
context = _template_env(request)
|
|
context.update(
|
|
bike=bike,
|
|
summary=summary,
|
|
events=events,
|
|
csrf=_csrf(request),
|
|
event_types=s.event_types,
|
|
)
|
|
return templates.TemplateResponse(request, "bikes/detail.html", context)
|
|
|
|
|
|
@app.get("/bikes/{bike_id}/edit", include_in_schema=False)
|
|
async def bike_edit_get(request: Request, bike_id: int):
|
|
require_user(request)
|
|
with d.get_conn() as conn:
|
|
bike = d.get_bike(conn, bike_id)
|
|
if bike is None:
|
|
return _redirect("/")
|
|
return _bike_render(request, "edit", bike, {}, {})
|
|
|
|
|
|
@app.post("/bikes/{bike_id}/edit", include_in_schema=False)
|
|
async def bike_edit_post(request: Request, bike_id: int):
|
|
require_user(request)
|
|
form = await request.form()
|
|
with d.get_conn() as conn:
|
|
bike = d.get_bike(conn, bike_id)
|
|
if bike is None:
|
|
return _redirect("/")
|
|
if not _csrf_ok(request, form):
|
|
errs = {"_": "CSRF check failed."}
|
|
return _bike_render(request, "edit", bike, errs, _form_dict(form))
|
|
try:
|
|
bike_in = _bike_in(form)
|
|
except Exception as exc:
|
|
return _bike_render(request, "edit", bike, _errors_from(exc), _form_dict(form))
|
|
d.update_bike(conn, bike_id, bike_in.name, bike_in.service_interval_miles)
|
|
return _redirect(f"/bikes/{bike_id}")
|
|
|
|
|
|
@app.post("/bikes/{bike_id}/delete", include_in_schema=False)
|
|
async def bike_delete(request: Request, bike_id: int):
|
|
s = get_settings()
|
|
token = request.cookies.get(SESSION_COOKIE)
|
|
form = await request.form()
|
|
if not check_csrf(s.secret_key, token, _form_str(form, "csrf")):
|
|
return _redirect(f"/bikes/{bike_id}")
|
|
with d.get_conn() as conn:
|
|
d.delete_bike(conn, bike_id)
|
|
return _redirect("/", status.HTTP_303_SEE_OTHER)
|
|
|
|
|
|
# --- events ------------------------------------------------------
|
|
@app.post("/bikes/{bike_id}/events", include_in_schema=False)
|
|
async def event_create(request: Request, bike_id: int):
|
|
"""Create an event; return the service card + fresh list (HTMX swap)."""
|
|
require_user(request)
|
|
s = get_settings()
|
|
form = await request.form()
|
|
|
|
if not _csrf_ok(request, form):
|
|
return _render_bike_partial(request, s, bike_id, errors={"_": "CSRF check failed."})
|
|
|
|
try:
|
|
event_in = _event_in(form)
|
|
except Exception as exc:
|
|
errors = _errors_from(exc)
|
|
return _render_bike_partial(request, s, bike_id, errors=errors, old=_form_dict(form))
|
|
|
|
with d.get_conn() as conn:
|
|
if d.get_bike(conn, bike_id) is None:
|
|
return _redirect("/")
|
|
d.create_event(
|
|
conn,
|
|
bike_id,
|
|
event_in.type,
|
|
event_in.event_date,
|
|
event_in.mileage,
|
|
event_in.cost,
|
|
event_in.notes,
|
|
)
|
|
return _render_bike_partial(request, s, bike_id)
|
|
|
|
|
|
def _render_bike_partial(
|
|
request, s: Settings, bike_id: int, errors: dict | None = None, old: dict | None = None
|
|
) -> Response:
|
|
"""Render the service card + fresh event list (HTMX swap into #bike-body)."""
|
|
base: dict = {
|
|
"request": request,
|
|
"today": date.today(),
|
|
"event_types": s.event_types,
|
|
"settings": s,
|
|
"bike_id": bike_id,
|
|
"csrf": csrf_token(s.secret_key, request.cookies.get(SESSION_COOKIE)),
|
|
}
|
|
with d.get_conn() as conn:
|
|
bike = d.get_bike(conn, bike_id) or d.Bike(bike_id, "?", None)
|
|
summary = svc.summarize_bike(conn, bike)
|
|
events = d.list_events(conn, bike.id)
|
|
base.update(bike=bike, summary=summary, events=events, old=old or {}, errors=errors or {})
|
|
return templates.TemplateResponse(request, "events/partial.html", base)
|
|
|
|
|
|
@app.post("/events/{event_id}/delete", include_in_schema=False)
|
|
async def event_delete(request: Request, event_id: int):
|
|
s = get_settings()
|
|
token = request.cookies.get(SESSION_COOKIE)
|
|
form = await request.form()
|
|
if not check_csrf(s.secret_key, token, _form_str(form, "csrf")):
|
|
return _redirect("/")
|
|
with d.get_conn() as conn:
|
|
r = conn.execute("SELECT bike_id FROM events WHERE id = ?", (event_id,)).fetchone()
|
|
if r is None:
|
|
return _redirect("/")
|
|
bike_id = r["bike_id"]
|
|
d.delete_event(conn, event_id)
|
|
return _redirect(f"/bikes/{bike_id}")
|