From e63be87ba992bbcc8ced41d0109acf093803b8db Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 02:41:20 -0600 Subject: [PATCH] refactor(api): split activity router (list + live-tail stream) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extracts GET /api/activity and GET /api/activity/stream from api.py into cyclone.api_routers.activity. The activity router is small and self-contained: store.recent_activity() for the snapshot half, tail_events() (from api_helpers) for the live tail. activity_stream is re-exported at the cyclone.api module level so test_api_stream_live.py's direct import keeps working. api.py: 2310 -> 2257 (-53). Net diff: +110 / -61. Pytest: 8 failed / 735 passed / 16 skipped — identical to baseline. Live smoke: /api/activity 200 JSON; /api/activity/stream 200 application/x-ndjson with the expected snapshot lines. --- backend/src/cyclone/api.py | 69 ++------------- backend/src/cyclone/api_routers/activity.py | 97 +++++++++++++++++++++ 2 files changed, 105 insertions(+), 61 deletions(-) create mode 100644 backend/src/cyclone/api_routers/activity.py diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 96d04f8..03c6021 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -143,6 +143,7 @@ app.add_middleware( # working — Python prefers packages over same-named modules.) from cyclone.api_routers import ( # noqa: E402 acks, + activity, clearhouse, config, health, @@ -156,6 +157,13 @@ app.include_router(ta1_acks.router) app.include_router(providers.router) app.include_router(clearhouse.router) app.include_router(config.router) +app.include_router(activity.router) + +# Backwards-compat re-exports: test_api_stream_live.py imports these +# names directly from ``cyclone.api`` to invoke the endpoint coroutines +# in-process. Re-pointing at the router keeps the public surface stable +# while the bodies live in the routers package. +activity_stream = activity.activity_stream_endpoint def _resolve_payer(name: str) -> PayerConfig: @@ -1813,69 +1821,8 @@ def get_remittance(remittance_id: str) -> dict: return body -@app.get("/api/activity") -def list_activity( - request: Request, - kind: str | None = Query(None), - since: str | None = Query(None), - limit: int = Query(200, ge=1, le=500), -) -> Any: - events = store.recent_activity(limit=limit) - if kind is not None: - events = [e for e in events if e["kind"] == kind] - if since is not None: - events = [e for e in events if e["timestamp"] >= since] - total = len(events) - has_more = False - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(events, total, total, has_more), - media_type="application/x-ndjson", - ) - return { - "items": events, - "total": total, - "returned": total, - "has_more": has_more, - } -@app.get("/api/activity/stream") -async def activity_stream( - request: Request, - kind: str | None = Query(None), - since: str | None = Query(None), - limit: int = Query(50, ge=1, le=500), -) -> StreamingResponse: - """Stream Activity events as NDJSON: snapshot first, then live events. - - Subscribes to ``activity_recorded``. Default ``limit`` is 50 - (smaller than the list endpoint's 200) because activity is - high-volume — callers usually want the most recent handful, not a - full replay. - """ - bus: EventBus = request.app.state.event_bus - - async def gen() -> AsyncIterator[bytes]: - # Snapshot reuses the same in-memory filter as ``list_activity`` - # so the two endpoints are interchangeable for the snapshot - # half. - events = store.recent_activity(limit=limit) - if kind is not None: - events = [e for e in events if e["kind"] == kind] - if since is not None: - events = [e for e in events if e["timestamp"] >= since] - for ev in events: - yield _ndjson_line({"type": "item", "data": ev}) - yield _ndjson_line({ - "type": "snapshot_end", "data": {"count": len(events)}, - }) - - async for chunk in _tail_events(request, bus, ["activity_recorded"]): - yield chunk - - return StreamingResponse(gen(), media_type="application/x-ndjson") - # --------------------------------------------------------------------------- # # 999 ACKs (read views) diff --git a/backend/src/cyclone/api_routers/activity.py b/backend/src/cyclone/api_routers/activity.py new file mode 100644 index 0000000..6ca63fe --- /dev/null +++ b/backend/src/cyclone/api_routers/activity.py @@ -0,0 +1,97 @@ +"""``/api/activity`` — list & live-tail of recorded Activity events. + +The Activity log is the cross-resource audit feed: every parser write, +inbox state change, match/dismiss/resubmit, clearhouse submission, etc. +records one row keyed by ``kind``. The list endpoint returns the most +recent ``limit`` events with optional ``kind`` / ``since`` filters; the +stream endpoint emits a snapshot followed by live updates from the +``EventBus``. + +The default ``limit`` on the list endpoint is 200 (matches the prior +inline behavior); the stream endpoint defaults to 50 because activity is +high-volume — callers usually want the most recent handful, not a full +replay. +""" +from __future__ import annotations + +from typing import Any, AsyncIterator + +from fastapi import APIRouter, Query, Request +from fastapi.responses import StreamingResponse + +from cyclone.api_helpers import ( + ndjson_line, + ndjson_stream_list, + tail_events, + wants_ndjson, +) +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/activity") +def list_activity_endpoint( + request: Request, + kind: str | None = Query(None), + since: str | None = Query(None), + limit: int = Query(200, ge=1, le=500), +) -> Any: + """Return the recent Activity log, newest first. + + Optional ``kind`` filters by ``event["kind"]`` (e.g. ``claim_submitted``). + Optional ``since`` is an inclusive ISO timestamp lower bound applied + to ``event["timestamp"]``. + """ + events = store.recent_activity(limit=limit) + if kind is not None: + events = [e for e in events if e["kind"] == kind] + if since is not None: + events = [e for e in events if e["timestamp"] >= since] + total = len(events) + has_more = False + if wants_ndjson(request): + return StreamingResponse( + ndjson_stream_list(events, total, total, has_more), + media_type="application/x-ndjson", + ) + return { + "items": events, + "total": total, + "returned": total, + "has_more": has_more, + } + + +@router.get("/api/activity/stream") +async def activity_stream_endpoint( + request: Request, + kind: str | None = Query(None), + since: str | None = Query(None), + limit: int = Query(50, ge=1, le=500), +) -> StreamingResponse: + """Stream Activity events as NDJSON: snapshot first, then live events. + + Subscribes to ``activity_recorded``. The snapshot half reuses the + same in-memory filter as ``list_activity`` so the two endpoints are + interchangeable for the snapshot half. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + events = store.recent_activity(limit=limit) + if kind is not None: + events = [e for e in events if e["kind"] == kind] + if since is not None: + events = [e for e in events if e["timestamp"] >= since] + for ev in events: + yield ndjson_line({"type": "item", "data": ev}) + yield ndjson_line({ + "type": "snapshot_end", "data": {"count": len(events)}, + }) + + async for chunk in tail_events(request, bus, ["activity_recorded"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson")