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())

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

TLS uses network.httpsServer (certFile / keyFile / optional caFile for mTLS).

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

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