Text & JSON Streams — Aquilia Documentation
Comprehensive guide and documentation for Text & JSON Streams in the Aquilia framework. View API reference, examples, and implementation patterns.
SSE SYSTEM / TEXT & JSON STREAMS Text & JSON Streaming Stream tokens or serializable payloads without wrapping elements manually. Leverage the SSEResponse.text() and SSEResponse.json() constructor overloads. SSEResponse.text() When streaming simple character files, LLM completion tokens, or raw console logs, wrapping strings inside SSEEvent objects is verbose. The SSEResponse.text() constructor accepts an AsyncGenerator[str, None]. It intercepts each string output from the generator, wraps it in a standard event payload, and writes it directly to the response socket. import asyncio from aquilia.controller import Controller, GET, RequestCtx from aquilia.sse import SSEResponse class LogStreamController(Controller): @GET("/sse/logs") async def stream_logs(self, ctx: RequestCtx): # Simply returns the text-based generator return SSEResponse.text(self._log_generator()) async def _log_generator(self): for line in ["Initialize boot...", "Load integrations...", "Compile routes..."]: yield line + "\\n" await asyncio.sleep(0.5) SSEResponse.json() For richer dashboards, updates must often be structured JSON payloads. Manual JSON encoding within generator functions adds boilerplates and risks runtime faults. The SSEResponse.json() constructor takes an AsyncGenerator[Any, None]. Each object yielded is encoded using json.dumps() and sent as an event. If serialization fails, the stream throws an SSESerializationFault . import asyncio from aquilia.controller import Controller, GET, RequestCtx from aquilia.sse import SSEResponse class IngestController(Controller): @GET("/sse/ingest/status") async def ingest_status(self, ctx: RequestCtx): # Returns the JSON object generator return SSEResponse.json(self._generate_status()) async def _generate_status(self): yield await asyncio.sleep(0.8) yield await asyncio.sleep(0.8) yield Standard Events OpenAI Streaming )
Go to Homepage