spb/company-atlas

1"""Live feed: `/live` (latest active events, never cached) and `/live/stream` (SSE)."""2from __future__ import annotations34from typing import Any56from fastapi import APIRouter, Query, Request, Response78from companyatlas.api import queries as q9from companyatlas.api import serializers as ser10from companyatlas.api.common import NO_STORE11from companyatlas.api.sse import MAX_STREAM_S, live_event_stream, sse_response12from companyatlas.db import connection1314ORDER = 1015router = APIRouter(prefix="/api/v1", tags=["live"])161718@router.get("/live", summary="Latest active events")19async def live(response: Response, limit: int = Query(50, ge=1, le=200), since: str | None = None, event_type: str | None = None,20               min_importance: float | None = Query(None, ge=0, le=1), country: str | None = None, industry: str | None = None) -> dict[str, Any]:21    response.headers["cache-control"] = NO_STORE22    since_dt = q.parse_iso(since)23    where, params = q.event_filters(event_type=event_type, min_importance=min_importance, country=country, industry=industry)24    if since_dt is not None:25        where.append("e.detected_at > :live_since")26        params["live_since"] = since_dt27    async with connection() as conn:28        rows = await q.fetch_events(conn, where, params, sort="recent", limit=limit)29    items = [ser.event(r) for r in rows]30    return {"items": items, "count": len(items), "cursor": items[0]["detected_at"] if items else (since_dt or q.now_utc()), "server_time": q.now_utc()}313233@router.get("/live/stream", summary="Server-sent events stream of new events")34async def live_stream(request: Request, since: str | None = None, event_type: str | None = None, min_importance: float | None = Query(None, ge=0, le=1),35                      max_s: float = Query(MAX_STREAM_S, ge=1, le=MAX_STREAM_S)) -> Any:36    since_dt = q.parse_iso(since)37    return sse_response(live_event_stream(request, since=since_dt, event_type=event_type, min_importance=min_importance, max_s=max_s))38