Network and SSE¶

HTTP server is FastAPI + uvicorn. SSE adds connection management, auth, and rate limiting.

HTTP server¶

Implement ITaskHandler and call server.run:

from fastapi import FastAPI

from pypepper.common.config import config
from pypepper.network.http import server
from pypepper.network.http.interfaces import ITaskHandler


class AppHandlers(ITaskHandler):
    def register_handlers(self, app: FastAPI):
        @app.get("/hello")
        def hello():
            return {"msg": "ok"}

    def use_middleware(self, app: FastAPI):
        pass


config.load_config("./conf/app.config.yaml")
server.run(AppHandlers())

server.run / run_without_tls / run_with_tls build a fresh app via create_app(...) (handlers + middleware registered once). Prefer create_app in tests instead of re-registering on the module-level server.app. server.run raises if both httpServer.enable and httpsServer.enable are true (enable only one; 0.7.0 breaking vs 0.6.6, which let HTTP win silently). build_response stays HTTP 200 by default; pass status_code explicitly (business code is not mapped to HTTP).

Built-in routes from BaseHandlers: /health, /ping, /metrics. RequestIdMiddleware injects X-Request-ID.

TLS uses network.httpsServer (certFile / keyFile). When mutualTLS is enabled, caFile is required and the server sets ssl_cert_reqs=ssl.CERT_REQUIRED (uvicorn would otherwise default to CERT_NONE even with ssl_ca_certs).

SSE handler¶

from collections.abc import AsyncIterator

from fastapi import FastAPI, Request
from fastapi.sse import EventSourceResponse

from pypepper.network.http.sse.event import SSEEvent
from pypepper.network.http.sse.interfaces import ISSEConnection, ISSEEvent, ISSEHandler
from pypepper.network.http.sse.security import require_sse_api_key
from pypepper.network.http.sse.stream import sse_stream


class ClockSSEHandler(ISSEHandler):
    async def on_connect(self, connection: ISSEConnection) -> None:
        await connection.send(SSEEvent(data={"status": "connected"}, event="connect"))

    async def on_disconnect(self, connection: ISSEConnection) -> None:
        return

    async def generate_events(self, connection: ISSEConnection) -> AsyncIterator[ISSEEvent]:
        yield SSEEvent(data={"tick": 1}, event="tick", id="1")


app = FastAPI()


@app.get("/sse/clock")
@require_sse_api_key
async def clock(request: Request):
    return EventSourceResponse(sse_stream(request, ClockSSEHandler()))

Auth and limits¶

Headers only: X-API-Key or Authorization: Bearer <key>
Query-string api_key is rejected
Keys come from YAML sse.authentication.validKeys (default empty — inject via deployment)
Rate limit: sse.rateLimit.maxRequestsPerMinute
Connection caps: maxTotalConnections, maxConnectionsPerIP
Reconnect: events with an id are kept in a bounded in-process ring (128). A new connection that sends Last-Event-ID is replayed events strictly after that id. Events without id are not stored. If the id is not in the ring, replay is empty and the stream continues live (this is not a complete bus).

Security notes¶

Production: keep sse.authentication.enabled: true and inject validKeys (do not commit real keys).
Auth off is blocked by default. Setting authentication.enabled: false rejects requests with 503 unless you explicitly export PYPEPPER_SSE_ALLOW_AUTH_OFF=1 (or true / yes / on). With that escape set, any non-empty API key is accepted and the library logs a one-shot warning (once per process).
Rate limiting is not a security boundary when auth is off. Limits are bucketed by the presented key string, so clients can rotate keys to bypass quotas. Treat rate limits as abuse soft-control only when authentication is enabled. Rate-limit counters live on the process-wide sse_security singleton (not a security upgrade; still per-process).

See the full working app in example/sse/app.py.

export PYPEPPER_SSE_API_KEY=your-local-key
python example/sse/app.py
curl -N -H "X-API-Key: $PYPEPPER_SSE_API_KEY" http://localhost:55550/sse/echo