component_framework.adapters.fastapi

  1"""FastAPI adapter for component endpoints."""
  2
  3import json
  4import logging
  5
  6try:
  7    from fastapi import HTTPException, Request
  8    from fastapi.responses import JSONResponse
  9    from starlette.responses import StreamingResponse
 10except ImportError as e:
 11    from . import _require_extra
 12
 13    raise _require_extra("fastapi", "fastapi") from e
 14
 15from ..core import StateSerializer, registry
 16from ..core.streaming import StreamingComponent, format_sse_frame
 17
 18logger = logging.getLogger(__name__)
 19
 20
 21async def component_endpoint(name: str, request: Request) -> JSONResponse:
 22    """
 23    Generic component endpoint for FastAPI.
 24
 25    POST /components/{name}
 26    Body: {
 27        "event": "event_name",
 28        "payload": {...},
 29        "state": "serialized_state"
 30    }
 31
 32    Returns: {
 33        "html": "rendered_html",
 34        "state": "serialized_state",
 35        "component_id": "component-id"
 36    }
 37    """
 38    try:
 39        # Get component class
 40        component_cls = registry.get(name)
 41        if not component_cls:
 42            raise HTTPException(status_code=404, detail=f"Component '{name}' not found")
 43
 44        # Parse request data
 45        try:
 46            data = await request.json()
 47        except Exception as e:
 48            raise HTTPException(status_code=400, detail=f"Invalid JSON: {e}")
 49
 50        # Extract parameters
 51        params = data.get("params", {})
 52        event = data.get("event")
 53        payload_raw = data.get("payload", {})
 54        state_str = data.get("state")
 55
 56        # Guard against double-serialised payload from older client JS
 57        if isinstance(payload_raw, str):
 58            try:
 59                payload = json.loads(payload_raw)
 60            except (json.JSONDecodeError, ValueError):
 61                payload = {}
 62        else:
 63            payload = payload_raw
 64
 65        # Deserialize state if provided
 66        state = None
 67        if state_str:
 68            try:
 69                state = StateSerializer.deserialize(state_str)
 70            except Exception as e:
 71                raise HTTPException(status_code=400, detail=f"Invalid state: {e}")
 72
 73        # Create and dispatch component (async to support async on_* handlers)
 74        component = component_cls(**params)
 75        result = await component.async_dispatch(event=event, payload=payload, state=state)
 76
 77        # Serialize state for response
 78        result["state"] = StateSerializer.serialize(result["state"])
 79
 80        return JSONResponse(content=result)
 81
 82    except HTTPException:
 83        raise
 84    except Exception:
 85        logger.exception(f"Error processing component '{name}'")
 86        raise HTTPException(status_code=500, detail="Internal server error")
 87
 88
 89async def stream_component_endpoint(name: str, request: Request) -> StreamingResponse:
 90    """
 91    SSE streaming endpoint for long-running component operations.
 92
 93    POST /components/{name}/stream
 94    Body: same as component_endpoint
 95
 96    Returns: text/event-stream with one ``data:`` frame per intermediate render.
 97    """
 98    try:
 99        component_cls = registry.get(name)
100        if not component_cls:
101            raise HTTPException(status_code=404, detail=f"Component '{name}' not found")
102
103        if not issubclass(component_cls, StreamingComponent):
104            raise HTTPException(
105                status_code=400,
106                detail=f"Component '{name}' does not support streaming",
107            )
108
109        try:
110            data = await request.json()
111        except Exception as e:
112            raise HTTPException(status_code=400, detail=f"Invalid JSON: {e}")
113
114        params = data.get("params", {})
115        event = data.get("event")
116        payload_raw = data.get("payload", {})
117        state_str = data.get("state")
118
119        if isinstance(payload_raw, str):
120            try:
121                payload = json.loads(payload_raw)
122            except (json.JSONDecodeError, ValueError):
123                payload = {}
124        else:
125            payload = payload_raw
126
127        state = None
128        if state_str:
129            try:
130                state = StateSerializer.deserialize(state_str)
131            except Exception as e:
132                raise HTTPException(status_code=400, detail=f"Invalid state: {e}")
133
134        component = component_cls(**params)
135
136        async def event_generator():
137            async for frame in component.async_stream_dispatch(
138                event=event, payload=payload, state=state
139            ):
140                frame["state"] = StateSerializer.serialize(frame["state"])
141                yield format_sse_frame(frame)
142
143        return StreamingResponse(event_generator(), media_type="text/event-stream")
144
145    except HTTPException:
146        raise
147    except Exception:
148        logger.exception(f"Error processing streaming component '{name}'")
149        raise HTTPException(status_code=500, detail="Internal server error")
150
151
152def create_component_routes(app):
153    """
154    Add component endpoints to FastAPI app.
155
156    Registers both the standard POST endpoint and the SSE streaming endpoint.
157
158    Usage:
159        from fastapi import FastAPI
160        app = FastAPI()
161        create_component_routes(app)
162    """
163    app.add_api_route(
164        "/components/{name}",
165        component_endpoint,
166        methods=["POST"],
167        name="component_endpoint",
168    )
169    app.add_api_route(
170        "/components/{name}/stream",
171        stream_component_endpoint,
172        methods=["POST"],
173        name="stream_component_endpoint",
174    )