refactor(api): split activity router (list + live-tail stream)

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.
This commit is contained in:
Tyler
2026-06-21 02:41:20 -06:00
parent 6aa440d3e4
commit e63be87ba9
2 changed files with 105 additions and 61 deletions
+8 -61
View File
@@ -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)
@@ -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")