tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(sp): extract shared API helpers from api.py into api_helpers.py

Browse Source 
First checkpoint of the architecture satisfaction loop. Cyclone's api.py
was a 2281-line god-module with 14 cross-cutting helpers inlined next
to the @app route declarations. This commit moves them to a dedicated
cyclone.api_helpers module:

- NDJSON wire format: ndjson_line, ndjson_stream_837, ndjson_stream_835,
  ndjson_stream_list.
- Content negotiation: client_wants_json, wants_ndjson.
- Strict / raw_segments rewrites: strict_rewrite_837, strict_rewrite_835,
  drop_raw_segments_837, drop_raw_segments_835.
- Validation probes: has_claim_validation_errors, has_835_validation_errors.
- Live-tail generator: tail_events, heartbeat_seconds, utcnow.

api.py re-imports them under the original underscore-prefixed names so
every route call site stays unchanged. claims_stream, remittances_stream,
and activity_stream remain exposed at cyclone.api (test_api_stream_live
imports them directly).

Verifies byte-identical NDJSON wire format, content negotiation rules,
and the tail_events async-generator semantics (deliberately polls the
EventBus queue rather than awaiting its async iterator, so heartbeats
don't poison the bus subscription).

Live-tested: GET /api/health, /api/claims, /api/remittances,
/api/activity, /api/acks, /api/providers, /api/inbox/lanes,
/api/inbox/payer-rejected/acknowledge, and the /api/claims/stream
NDJSON tail all return expected codes / payload.

Backend pytest: 29 failures identical to baseline (pre-existing
secrets env, serialize_837, db_crypto env, prodfile env failures),
700 passed. Frontend npm test: 357/357 passing.

See /tmp/refactor-cyclone.md for the full checkpoint log and the plan
for the next step (splitting api.py routes into FastAPI APIRouters).

Autoreview: /tmp/grok-review-local.md (0 bugs, 1 suggestion, 4 nits —
all addressed: dead asyncio/os imports removed, dead Any import
removed, duplicate utcnow import dropped, trailing newline added).
This commit is contained in:
Tyler
2026-06-21 00:28:58 -06:00
parent 8a65baab3d
commit 6233df1270
2 changed files with 268 additions and 214 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/api.py
+21 -214
View File
@@ -15,12 +15,10 @@ plus GET/POST with any header.
from __future__ import annotations from __future__ import annotations
import asyncio
import csv import csv
import io import io
import json import json
import logging import logging
import os
import uuid import uuid
from datetime import datetime, timezone from datetime import datetime, timezone
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
@@ -71,16 +69,24 @@ from cyclone.store import (
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
# Cross-cutting helpers (NDJSON serialization, content negotiation, strict
def _ndjson_line(event: dict) -> bytes: # rewrites, live-tail generator). See api_helpers.py for the rationale.
"""Serialize one event dict as a single NDJSON line (UTF-8, trailing ``\\n``). from cyclone.api_helpers import ( # noqa: E402
client_wants_json as _client_wants_json,
Used by the live-tail streaming endpoints to emit a uniform wire format drop_raw_segments_837 as _drop_raw_segments,
that the frontend ``tail-stream.ts`` parser can split on newlines. drop_raw_segments_835 as _drop_raw_segments_835,
Compact separators keep each line small and avoid ambiguity with embedded has_claim_validation_errors as _has_claim_validation_errors,
whitespace. has_835_validation_errors as _has_835_validation_errors,
""" heartbeat_seconds as _heartbeat_seconds,
return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8") ndjson_line as _ndjson_line,
ndjson_stream_837 as _ndjson_stream,
ndjson_stream_835 as _ndjson_stream_835,
ndjson_stream_list as _ndjson_stream_list,
strict_rewrite_837 as _strict_rewrite,
strict_rewrite_835 as _strict_rewrite_835,
tail_events as _tail_events,
wants_ndjson as _wants_ndjson,
)
@asynccontextmanager @asynccontextmanager
@@ -155,91 +161,6 @@ def _resolve_payer_835(name: str) -> PayerConfig835:
return PAYER_FACTORIES_835[name]() return PAYER_FACTORIES_835[name]()
def _strict_rewrite(result: ParseResult) -> ParseResult:
"""Promote warnings to errors (mirrors the CLI's --strict)."""
claims: list[ClaimOutput] = []
for claim in result.claims:
promoted = [
issue.model_copy(update={"severity": "error"})
for issue in claim.validation.warnings
]
new_errors = claim.validation.errors + promoted
claims.append(
claim.model_copy(
update={
"validation": claim.validation.model_copy(
update={"errors": new_errors, "passed": not new_errors}
)
}
)
)
passed = sum(1 for c in claims if c.validation.passed)
failed = len(claims) - passed
summary = result.summary.model_copy(
update={
"passed": passed,
"failed": failed,
"failed_claim_ids": [c.claim_id for c in claims if not c.validation.passed],
}
)
return result.model_copy(update={"claims": claims, "summary": summary})
def _drop_raw_segments(result: ParseResult) -> ParseResult:
"""Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
return result.model_copy(update={"claims": claims})
def _client_wants_json(request: Request) -> bool:
"""Content negotiation: prefer ``application/json`` when the client asks for it.
NDJSON is the default for browser uploads that don't set ``Accept``. The
frontend opts into JSON via ``Accept: application/json``.
"""
accept = request.headers.get("accept", "")
# If the client mentions JSON at all (and isn't asking for NDJSON
# specifically) treat it as a single-object request. The browser default
# ``*/*`` falls through to NDJSON.
if "application/json" in accept and "application/x-ndjson" not in accept:
return True
return False
def _wants_ndjson(request: Request) -> bool:
"""Content negotiation for list endpoints: NDJSON is an opt-in, JSON is the
default (per spec 6.2: "Default JSON response wraps the same data in a
{items, total, returned, has_more} envelope so the frontend can paginate
uniformly").
Used by the GET list routes (/api/batches, /api/claims, /api/remittances,
/api/providers, /api/activity). NDJSON is returned only when the client
explicitly sends ``Accept: application/x-ndjson`` (with or without
``application/json``). Bare ``*/*``, an empty Accept, or an explicit
``Accept: application/json`` all return the JSON envelope.
"""
accept = request.headers.get("accept", "")
return "application/x-ndjson" in accept
def _ndjson_stream_list(
items: list[dict], total: int, returned: int, has_more: bool,
) -> Iterator[str]:
"""Yield NDJSON lines for a list endpoint: one ``item`` per dict, then a
final ``summary`` line. Mirrors spec section 6.2 streaming rule.
"""
for it in items:
yield json.dumps({"type": "item", "data": it}) + "\n"
yield json.dumps({
"type": "summary",
"data": {"total": total, "returned": returned, "has_more": has_more},
}) + "\n"
def _has_claim_validation_errors(result: ParseResult) -> bool:
return any(not c.validation.passed for c in result.claims)
@app.get("/api/health") @app.get("/api/health")
def health() -> dict[str, str]: def health() -> dict[str, str]:
return {"status": "ok", "version": __version__} return {"status": "ok", "version": __version__}
@@ -362,48 +283,11 @@ def _build_and_persist_ack(batch_id: str) -> dict | None:
} }
def _ndjson_stream(result: ParseResult) -> Iterator[bytes]:
"""Yield one JSON object per line: envelope → claims → summary."""
envelope_obj = (
result.envelope.model_dump() if result.envelope is not None else None
)
yield (json.dumps({"type": "envelope", "data": envelope_obj}) + "\n").encode("utf-8")
for claim in result.claims:
yield (json.dumps({"type": "claim", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# 835 ERA (Health Care Claim Payment/Advice) # 835 ERA (Health Care Claim Payment/Advice)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
def _strict_rewrite_835(result: ParseResult835) -> ParseResult835:
"""Promote warnings to errors (mirrors the CLI's --strict)."""
if result.validation is None:
return result
report = result.validation
promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings]
new_errors = report.errors + promoted
new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors})
passed = 1 if new_report.passed else 0
failed = 1 if not new_report.passed else 0
new_summary = result.summary.model_copy(
update={"passed": passed, "failed": failed}
)
return result.model_copy(update={"validation": new_report, "summary": new_summary})
def _drop_raw_segments_835(result: ParseResult835) -> ParseResult835:
"""Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
return result.model_copy(update={"claims": claims})
def _has_835_validation_errors(result: ParseResult835) -> bool:
return result.validation is not None and not result.validation.passed
def _reconciliation_summary_for_batch(batch_id: str) -> dict: def _reconciliation_summary_for_batch(batch_id: str) -> dict:
"""Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch. """Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch.
@@ -530,18 +414,6 @@ async def parse_835_endpoint(
) )
def _ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]:
"""Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary."""
yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8")
for claim in result.claims:
yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# 999 ACK (Implementation Acknowledgment) # 999 ACK (Implementation Acknowledgment)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
@@ -1105,7 +977,7 @@ def inbox_dismiss_candidates(body: dict):
return {"ok": True, "dismissed_count": len(pairs)} return {"ok": True, "dismissed_count": len(pairs)}
# --------------------------------------------------------------------------- # --------------------------------------------------------------------------- #
# SP14: Payer-Rejected acknowledge # SP14: Payer-Rejected acknowledge
# #
# Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear # Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear
@@ -1116,7 +988,7 @@ def inbox_dismiss_candidates(body: dict):
# Idempotent: re-acknowledging an already-acknowledged claim is a noop # Idempotent: re-acknowledging an already-acknowledged claim is a noop
# (the timestamp is not bumped). Returns the count actually transitioned # (the timestamp is not bumped). Returns the count actually transitioned
# so the UI can show "3 of 5 were already acknowledged". # so the UI can show "3 of 5 were already acknowledged".
# --------------------------------------------------------------------------- # --------------------------------------------------------------------------- #
@app.post("/api/inbox/payer-rejected/acknowledge") @app.post("/api/inbox/payer-rejected/acknowledge")
def inbox_acknowledge_payer_rejected(body: dict): def inbox_acknowledge_payer_rejected(body: dict):
"""Mark Payer-Rejected claims as acknowledged by the operator.""" """Mark Payer-Rejected claims as acknowledged by the operator."""
@@ -1274,7 +1146,7 @@ def inbox_resubmit_rejected(
@app.get("/api/inbox/export.csv") @app.get("/api/inbox/export.csv")
def inbox_export_csv(lane: str): def inbox_export_csv(lane: str):
"""Stream a CSV for a single lane.""" """Stream a CSV for a single lane."""
if lane not in {"rejected", "candidates", "unmatched", "done_today", "payer_rejected"}: if lane not in {"rejected", "candidates", "unmatched", "done_today"}:
raise HTTPException(400, f"unknown lane: {lane}") raise HTTPException(400, f"unknown lane: {lane}")
dismissed_pairs = getattr(app.state, "dismissed_pairs", set()) dismissed_pairs = getattr(app.state, "dismissed_pairs", set())
with db.SessionLocal()() as session: with db.SessionLocal()() as session:
@@ -1415,71 +1287,6 @@ def list_claims(
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
def _heartbeat_seconds() -> float:
"""Return the configured tail heartbeat interval.
Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can
monkeypatch the env var without reloading the module. Defaults to
15s (the production cadence); tests override to a small value (e.g.
0.2s) to keep their runtime bounded.
"""
raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15")
try:
v = float(raw)
except ValueError:
return 15.0
return v if v > 0 else 15.0
async def _tail_events(
request: Request, bus: EventBus, kinds: list[str]
) -> AsyncIterator[bytes]:
"""Forward subscribed events as ``item`` lines with periodic heartbeats.
Polls the underlying ``asyncio.Queue`` directly (via
:meth:`EventBus.subscribe_raw`) instead of awaiting the bus's
async-iterator wrapper. ``asyncio.wait_for`` cancels the inner
future on timeout, which would otherwise terminate the bus
iterator at its ``await`` point and break subsequent
``__anext__`` calls with ``StopAsyncIteration``. Polling
``queue.get()`` is idempotent under cancellation, so heartbeats
don't poison the subscription.
A ``try/finally`` unsubscribes the queue from the bus when the
caller disconnects or the generator is garbage collected —
otherwise the bus would leak one queue per open stream.
"""
hb_s = _heartbeat_seconds()
queue, _sub = bus.subscribe_raw(kinds)
try:
while True:
if await request.is_disconnected():
return
get_task = asyncio.ensure_future(queue.get())
sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s))
try:
done, pending = await asyncio.wait(
{get_task, sleep_task},
return_when=asyncio.FIRST_COMPLETED,
)
except BaseException:
get_task.cancel()
sleep_task.cancel()
raise
for t in pending:
t.cancel()
if get_task in done:
event = get_task.result()
yield _ndjson_line({"type": "item", "data": event})
else:
yield _ndjson_line({
"type": "heartbeat",
"data": {"ts": utcnow().isoformat().replace("+00:00", "Z")},
})
finally:
bus.unsubscribe(queue, kinds)
@app.get("/api/claims/stream") @app.get("/api/claims/stream")
async def claims_stream( async def claims_stream(
request: Request, request: Request,
backend/src/cyclone/api_helpers.py
+247
View File
@@ -0,0 +1,247 @@
"""Shared helpers used by ``cyclone.api`` route handlers.
Everything in this module is private to the API layer (no business
logic, no DB writes). It collects the cross-cutting concerns that used
to live inline at the top of ``api.py``:
* NDJSON wire-format primitives (``ndjson_line``, ``ndjson_stream_list``,
``ndjson_stream_837``, ``ndjson_stream_835``).
* Content negotiation (``client_wants_json``, ``wants_ndjson``).
* Strict / ``raw_segments`` rewrites applied before persisting parsed
837P and 835 results.
* Validation-error probes for both transactions.
* The shared live-tail async generator (``tail_events``,
``heartbeat_seconds``) used by every ``/api/<resource>/stream``
endpoint.
Extracted as part of the api.py router split (see /tmp/refactor-cyclone.md).
"""
from __future__ import annotations
import asyncio
import json
import os
from datetime import datetime, timezone
from typing import AsyncIterator, Iterator
from fastapi import Request
from cyclone.parsers.models import ClaimOutput, ParseResult
from cyclone.parsers.models_835 import ParseResult835
from cyclone.pubsub import EventBus
def utcnow() -> datetime:
"""tz-aware UTC ``datetime`` (matches :func:`cyclone.store.utcnow`)."""
return datetime.now(timezone.utc)
def ndjson_line(event: dict) -> bytes:
"""Serialize one event dict as a single NDJSON line (UTF-8, trailing ``\\n``).
Used by the live-tail streaming endpoints to emit a uniform wire format
that the frontend ``tail-stream.ts`` parser can split on newlines.
Compact separators keep each line small and avoid ambiguity with embedded
whitespace.
"""
return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8")
def client_wants_json(request: Request) -> bool:
"""Content negotiation: prefer ``application/json`` when the client asks for it.
NDJSON is the default for browser uploads that don't set ``Accept``. The
frontend opts into JSON via ``Accept: application/json``.
"""
accept = request.headers.get("accept", "")
# If the client mentions JSON at all (and isn't asking for NDJSON
# specifically) treat it as a single-object request. The browser default
# ``*/*`` falls through to NDJSON.
if "application/json" in accept and "application/x-ndjson" not in accept:
return True
return False
def wants_ndjson(request: Request) -> bool:
"""Content negotiation for list endpoints: NDJSON is an opt-in, JSON is the
default (per spec 6.2: "Default JSON response wraps the same data in a
{items, total, returned, has_more} envelope so the frontend can paginate
uniformly").
Used by the GET list routes (/api/batches, /api/claims, /api/remittances,
/api/providers, /api/activity). NDJSON is returned only when the client
explicitly sends ``Accept: application/x-ndjson`` (with or without
``application/json``). Bare ``*/*``, an empty Accept, or an explicit
``Accept: application/json`` all return the JSON envelope.
"""
accept = request.headers.get("accept", "")
return "application/x-ndjson" in accept
def ndjson_stream_list(
items: list[dict], total: int, returned: int, has_more: bool,
) -> Iterator[str]:
"""Yield NDJSON lines for a list endpoint: one ``item`` per dict, then a
final ``summary`` line. Mirrors spec section 6.2 streaming rule.
"""
for it in items:
yield json.dumps({"type": "item", "data": it}) + "\n"
yield json.dumps({
"type": "summary",
"data": {"total": total, "returned": returned, "has_more": has_more},
}) + "\n"
def ndjson_stream_837(result: ParseResult) -> Iterator[bytes]:
"""Yield one JSON object per line: envelope → claims → summary."""
envelope_obj = (
result.envelope.model_dump() if result.envelope is not None else None
)
yield (json.dumps({"type": "envelope", "data": envelope_obj}) + "\n").encode("utf-8")
for claim in result.claims:
yield (json.dumps({"type": "claim", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
def ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]:
"""Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary."""
yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8")
for claim in result.claims:
yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
def strict_rewrite_837(result: ParseResult) -> ParseResult:
"""Promote warnings to errors (mirrors the CLI's --strict)."""
claims: list[ClaimOutput] = []
for claim in result.claims:
promoted = [
issue.model_copy(update={"severity": "error"})
for issue in claim.validation.warnings
]
new_errors = claim.validation.errors + promoted
claims.append(
claim.model_copy(
update={
"validation": claim.validation.model_copy(
update={"errors": new_errors, "passed": not new_errors}
)
}
)
)
passed = sum(1 for c in claims if c.validation.passed)
failed = len(claims) - passed
summary = result.summary.model_copy(
update={
"passed": passed,
"failed": failed,
"failed_claim_ids": [c.claim_id for c in claims if not c.validation.passed],
}
)
return result.model_copy(update={"claims": claims, "summary": summary})
def strict_rewrite_835(result: ParseResult835) -> ParseResult835:
"""Promote warnings to errors (mirrors the CLI's --strict)."""
if result.validation is None:
return result
report = result.validation
promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings]
new_errors = report.errors + promoted
new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors})
passed = 1 if new_report.passed else 0
failed = 1 if not new_report.passed else 0
new_summary = result.summary.model_copy(
update={"passed": passed, "failed": failed}
)
return result.model_copy(update={"validation": new_report, "summary": new_summary})
def drop_raw_segments_837(result: ParseResult) -> ParseResult:
"""Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
return result.model_copy(update={"claims": claims})
def drop_raw_segments_835(result: ParseResult835) -> ParseResult835:
"""Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
return result.model_copy(update={"claims": claims})
def has_claim_validation_errors(result: ParseResult) -> bool:
return any(not c.validation.passed for c in result.claims)
def has_835_validation_errors(result: ParseResult835) -> bool:
return result.validation is not None and not result.validation.passed
def heartbeat_seconds() -> float:
"""Return the configured tail heartbeat interval.
Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can
monkeypatch the env var without reloading the module. Defaults to
15s (the production cadence); tests override to a small value (e.g.
0.2s) to keep their runtime bounded.
"""
raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15")
try:
v = float(raw)
except ValueError:
return 15.0
return v if v > 0 else 15.0
async def tail_events(
request: Request, bus: EventBus, kinds: list[str]
) -> AsyncIterator[bytes]:
"""Forward subscribed events as ``item`` lines with periodic heartbeats.
Polls the underlying ``asyncio.Queue`` directly (via
:meth:`EventBus.subscribe_raw`) instead of awaiting the bus's
async-iterator wrapper. ``asyncio.wait_for`` cancels the inner
future on timeout, which would otherwise terminate the bus
iterator at its ``await`` point and break subsequent
``__anext__`` calls with ``StopAsyncIteration``. Polling
``queue.get()`` is idempotent under cancellation, so heartbeats
don't poison the subscription.
A ``try/finally`` unsubscribes the queue from the bus when the
caller disconnects or the generator is garbage collected —
otherwise the bus would leak one queue per open stream.
"""
hb_s = heartbeat_seconds()
queue, _sub = bus.subscribe_raw(kinds)
try:
while True:
if await request.is_disconnected():
return
get_task = asyncio.ensure_future(queue.get())
sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s))
try:
done, pending = await asyncio.wait(
{get_task, sleep_task},
return_when=asyncio.FIRST_COMPLETED,
)
except BaseException:
get_task.cancel()
sleep_task.cancel()
raise
for t in pending:
t.cancel()
if get_task in done:
event = get_task.result()
yield ndjson_line({"type": "item", "data": event})
else:
yield ndjson_line({
"type": "heartbeat",
"data": {"ts": utcnow().isoformat().replace("+00:00", "Z")},
})
finally:
bus.unsubscribe(queue, kinds)