Skip to content

Network and SSE

FastAPI HTTP server and Server-Sent Events. Narrative guide: Network and SSE.

HTTP server

pypepper.network.http.server

FastAPI HTTP server run helpers (plain and TLS).

app module-attribute 

app = FastAPI()

run_without_tls 

run_without_tls(
    port: int,
    handlers_: ITaskHandler | None,
    host: str = "0.0.0.0",
) -> None
Source code in pypepper/network/http/server.py 
13
14
15
16
def run_without_tls(port: int, handlers_: ITaskHandler | None, host: str = "0.0.0.0") -> None:
    handlers.register_handlers(app, handlers_)
    handlers.use_middleware(app, handlers_)
    uvicorn.run(app, host=host, port=port, timeout_keep_alive=30)

run_with_tls 

run_with_tls(
    port: int,
    handlers_: ITaskHandler | None,
    host: str = "0.0.0.0",
) -> None
Source code in pypepper/network/http/server.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def run_with_tls(port: int, handlers_: ITaskHandler | None, host: str = "0.0.0.0") -> None:
    network_conf = config.get_yml_config().network
    https = network_conf.httpsServer
    cert_file = getattr(https, "certFile", None) or ""
    key_file = getattr(https, "keyFile", None) or ""
    ca_file = getattr(https, "caFile", None) or ""

    if not cert_file or not key_file:
        raise ValueError("HTTPS enabled but certFile/keyFile missing in network.httpsServer config")

    handlers.register_handlers(app, handlers_)
    handlers.use_middleware(app, handlers_)

    if getattr(https, "mutualTLS", False) and ca_file:
        uvicorn.run(
            app,
            host=host,
            port=port,
            timeout_keep_alive=30,
            ssl_certfile=cert_file,
            ssl_keyfile=key_file,
            ssl_ca_certs=ca_file,
        )
    else:
        uvicorn.run(
            app,
            host=host,
            port=port,
            timeout_keep_alive=30,
            ssl_certfile=cert_file,
            ssl_keyfile=key_file,
        )

run 

run(handlers_: ITaskHandler | None = None) -> None
Source code in pypepper/network/http/server.py 
53
54
55
56
57
58
59
60
61
def run(handlers_: ITaskHandler | None = None) -> None:
    network_conf = config.get_yml_config().network
    host = getattr(network_conf, "ip", None) or "0.0.0.0"
    if network_conf.httpServer.enable:
        run_without_tls(network_conf.httpServer.port, handlers_, host=host)
    elif network_conf.httpsServer.enable:
        run_with_tls(network_conf.httpsServer.port, handlers_, host=host)
    else:
        raise RuntimeError("Neither httpServer nor httpsServer is enabled in config")

HTTP interfaces

pypepper.network.http.interfaces

HTTP task handler and middleware interfaces.

ITaskHandler

Source code in pypepper/network/http/interfaces.py 
 8
 9
10
11
12
13
14
15
class ITaskHandler(metaclass=ABCMeta):
    @abstractmethod
    def register_handlers(self, app: FastAPI) -> None:
        pass

    @abstractmethod
    def use_middleware(self, app: FastAPI) -> None:
        pass

register_handlers abstractmethod 

register_handlers(app: FastAPI) -> None
Source code in pypepper/network/http/interfaces.py 
 9
10
11
@abstractmethod
def register_handlers(self, app: FastAPI) -> None:
    pass

use_middleware abstractmethod 

use_middleware(app: FastAPI) -> None
Source code in pypepper/network/http/interfaces.py 
13
14
15
@abstractmethod
def use_middleware(self, app: FastAPI) -> None:
    pass

HTTP handlers

pypepper.network.http.handlers.base

Built-in health, ping, and metrics handlers.

health async 

health(request: Request) -> JSONResponse
Source code in pypepper/network/http/handlers/base.py 
17
18
19
async def health(request: Request) -> JSONResponse:
    log.request_id().trace("Receive HealthCheck. URL.Path={}", request.url.path)
    return response.build_response(code="200", data=version.get_version_info(), msg="OK")

ping async 

ping() -> str
Source code in pypepper/network/http/handlers/base.py 
22
23
24
async def ping() -> str:
    log.request_id().debug("pong")
    return "pong"

metrics async 

metrics(request: Request) -> JSONResponse
Source code in pypepper/network/http/handlers/base.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
async def metrics(request: Request) -> JSONResponse:
    log.request_id().info("Receive MetricsCheck. URL.Path={}", request.url.path)
    data = {
        "uptime_seconds": round(time.time() - _PROCESS_START, 3),
        "version": version.get_version_info(),
    }
    try:
        from pypepper.network.http.sse.connection import connection_manager

        stats = connection_manager.get_stats()
        data["sse"] = {
            "total_connections": stats.get("total_connections", 0),
            "active_connections": stats.get("active_connections", 0),
            "total_dropped_events": stats.get("total_dropped_events", 0),
        }
    except Exception:
        pass
    return response.build_response(code="200", data=data, msg="OK")

pypepper.network.http.handlers.handlers

Default HTTP handler registration and middleware.

base_handlers module-attribute 

base_handlers = BaseHandlers()

RequestIdMiddleware

Bases: BaseHTTPMiddleware

Inject X-Request-ID and emit a basic access log line.

Source code in pypepper/network/http/handlers/handlers.py 
17
18
19
20
21
22
23
24
25
26
class RequestIdMiddleware(BaseHTTPMiddleware):
    """Inject X-Request-ID and emit a basic access log line."""

    async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
        req_id = request.headers.get("X-Request-ID") or uuid.new_uuid()
        request.state.request_id = req_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = req_id
        log.request_id(req_id).info(f"{request.method} {request.url.path} -> {response.status_code}")
        return response

dispatch async 

dispatch(
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response
Source code in pypepper/network/http/handlers/handlers.py 
20
21
22
23
24
25
26
async def dispatch(self, request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
    req_id = request.headers.get("X-Request-ID") or uuid.new_uuid()
    request.state.request_id = req_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = req_id
    log.request_id(req_id).info(f"{request.method} {request.url.path} -> {response.status_code}")
    return response

BaseHandlers

Bases: ITaskHandler

Source code in pypepper/network/http/handlers/handlers.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class BaseHandlers(ITaskHandler):
    def register_handlers(self, app: FastAPI) -> None:
        self._register_health_check(app)
        self._register_metrics_check(app)

    def use_middleware(self, app: FastAPI) -> None:
        self._use_default_middleware(app)

    @staticmethod
    def _register_health_check(app: FastAPI) -> None:
        app.get("/health")(health)
        app.get("/ping")(ping)

    @staticmethod
    def _register_metrics_check(app: FastAPI) -> None:
        app.get("/metrics")(metrics)

    def _use_default_middleware(self, app: FastAPI) -> None:
        # Starlette runs last-added middleware first. Tracing is outer; RequestId runs
        # inside the span so request_id is available after call_next.
        app.add_middleware(RequestIdMiddleware)
        from pypepper.common.tracing import TracingMiddleware

        app.add_middleware(TracingMiddleware)

register_handlers 

register_handlers(app: FastAPI) -> None
Source code in pypepper/network/http/handlers/handlers.py 
30
31
32
def register_handlers(self, app: FastAPI) -> None:
    self._register_health_check(app)
    self._register_metrics_check(app)

use_middleware 

use_middleware(app: FastAPI) -> None
Source code in pypepper/network/http/handlers/handlers.py 
34
35
def use_middleware(self, app: FastAPI) -> None:
    self._use_default_middleware(app)

register_handlers 

register_handlers(
    app: FastAPI,
    private_handlers: ITaskHandler | None = None,
) -> None
Source code in pypepper/network/http/handlers/handlers.py 
58
59
60
61
def register_handlers(app: FastAPI, private_handlers: ITaskHandler | None = None) -> None:
    base_handlers.register_handlers(app)
    if private_handlers:
        private_handlers.register_handlers(app)

use_middleware 

use_middleware(
    app: FastAPI,
    private_handlers: ITaskHandler | None = None,
) -> None
Source code in pypepper/network/http/handlers/handlers.py 
64
65
66
67
def use_middleware(app: FastAPI, private_handlers: ITaskHandler | None = None) -> None:
    base_handlers.use_middleware(app)
    if private_handlers:
        private_handlers.use_middleware(app)

SSE

pypepper.network.http.sse.connection

SSE connection registry and limits.

T module-attribute 

T = TypeVar('T')

connection_manager module-attribute 

connection_manager = SSEConnectionManager()

SSEConnection

Bases: ISSEConnection

SSE Connection implementation

Source code in pypepper/network/http/sse/connection.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class SSEConnection(ISSEConnection):
    """SSE Connection implementation"""

    DEFAULT_MAX_QUEUE_SIZE = 500

    def __init__(
        self,
        connection_id: str,
        queue: asyncio.Queue,
        context: Context | None = None,
        last_event_id: str | None = None,
    ):
        self.connection_id = connection_id
        self.context = context or Context(context_id=connection_id)
        self.last_event_id = last_event_id
        self._queue = queue
        self._closed = False
        self._dropped_events = 0

    @classmethod
    def max_queue_size(cls) -> int:
        return int(_sse_config_value("maxQueueSize", cls.DEFAULT_MAX_QUEUE_SIZE))

    async def send(self, event: ISSEEvent) -> bool:
        """
        Send event to client

        :param event: SSE event
        :return: True if sent successfully, False if dropped
        """
        if self._closed:
            return False

        try:
            # Non-blocking mode: drop event if queue is full
            self._queue.put_nowait(event)
            log.request_id(self.connection_id).debug(f"SSE event sent: event={event.event}, id={event.id}")
            return True

        except asyncio.QueueFull:
            self._dropped_events += 1
            log.request_id(self.connection_id).warn(
                f"SSE event dropped (queue full): event={event.event}, total_dropped={self._dropped_events}"
            )
            return False

    async def ping(self) -> None:
        """Send heartbeat to keep connection alive"""
        from pypepper.network.http.sse.event import SSEEvent

        await self.send(SSEEvent.ping())

    async def disconnect(self) -> None:
        """Disconnect and cleanup resources"""
        self._closed = True
        log.request_id(self.connection_id).info(f"SSE connection closed: {self.connection_id}")

    def is_closed(self) -> bool:
        """
        Check if connection is closed
        :return: True if closed
        """
        return self._closed

    def get_stats(self) -> dict:
        """
        Get connection statistics
        :return: Statistics dict
        """
        return {
            "connection_id": self.connection_id,
            "queue_size": self._queue.qsize(),
            "dropped_events": self._dropped_events,
            "is_closed": self._closed,
            "last_event_id": self.last_event_id,
        }

DEFAULT_MAX_QUEUE_SIZE class-attribute instance-attribute 

DEFAULT_MAX_QUEUE_SIZE = 500

connection_id instance-attribute 

connection_id = connection_id

context instance-attribute 

context = context or Context(context_id=connection_id)

last_event_id instance-attribute 

last_event_id = last_event_id

max_queue_size classmethod 

max_queue_size() -> int
Source code in pypepper/network/http/sse/connection.py 
52
53
54
@classmethod
def max_queue_size(cls) -> int:
    return int(_sse_config_value("maxQueueSize", cls.DEFAULT_MAX_QUEUE_SIZE))

send async 

send(event: ISSEEvent) -> bool

Send event to client

Parameters:

Name Type Description Default
event ISSEEvent

SSE event

 required

Returns:

Type Description
bool

True if sent successfully, False if dropped
Source code in pypepper/network/http/sse/connection.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
async def send(self, event: ISSEEvent) -> bool:
    """
    Send event to client

    :param event: SSE event
    :return: True if sent successfully, False if dropped
    """
    if self._closed:
        return False

    try:
        # Non-blocking mode: drop event if queue is full
        self._queue.put_nowait(event)
        log.request_id(self.connection_id).debug(f"SSE event sent: event={event.event}, id={event.id}")
        return True

    except asyncio.QueueFull:
        self._dropped_events += 1
        log.request_id(self.connection_id).warn(
            f"SSE event dropped (queue full): event={event.event}, total_dropped={self._dropped_events}"
        )
        return False

ping async 

ping() -> None

Send heartbeat to keep connection alive

Source code in pypepper/network/http/sse/connection.py 
79
80
81
82
83
async def ping(self) -> None:
    """Send heartbeat to keep connection alive"""
    from pypepper.network.http.sse.event import SSEEvent

    await self.send(SSEEvent.ping())

disconnect async 

disconnect() -> None

Disconnect and cleanup resources

Source code in pypepper/network/http/sse/connection.py 
85
86
87
88
async def disconnect(self) -> None:
    """Disconnect and cleanup resources"""
    self._closed = True
    log.request_id(self.connection_id).info(f"SSE connection closed: {self.connection_id}")

is_closed 

is_closed() -> bool

Check if connection is closed :return: True if closed

Source code in pypepper/network/http/sse/connection.py 
90
91
92
93
94
95
def is_closed(self) -> bool:
    """
    Check if connection is closed
    :return: True if closed
    """
    return self._closed

get_stats 

get_stats() -> dict

Get connection statistics :return: Statistics dict

Source code in pypepper/network/http/sse/connection.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
def get_stats(self) -> dict:
    """
    Get connection statistics
    :return: Statistics dict
    """
    return {
        "connection_id": self.connection_id,
        "queue_size": self._queue.qsize(),
        "dropped_events": self._dropped_events,
        "is_closed": self._closed,
        "last_event_id": self.last_event_id,
    }

SSEConnectionManager

Bases: ISSEConnectionManager

SSE Connection Manager (explicit singleton)

Source code in pypepper/network/http/sse/connection.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
class SSEConnectionManager(ISSEConnectionManager):
    """SSE Connection Manager (explicit singleton)"""

    DEFAULT_MAX_CONNECTIONS = 100
    DEFAULT_MAX_CONNECTIONS_PER_IP = 5

    _instance: SSEConnectionManager | None = None
    _init_lock = Lock()
    _lock: Lock
    _connections: MutableMapping[str, SSEConnection]

    def __new__(cls) -> SSEConnectionManager:
        with cls._init_lock:
            if cls._instance is None:
                inst = super().__new__(cls)
                inst._lock = Lock()
                inst._connections = {}
                cls._instance = inst
            return cls._instance

    def __init__(self) -> None:
        pass

    @property
    def MAX_CONNECTIONS(self) -> int:
        return self._effective_max_connections()

    @MAX_CONNECTIONS.setter
    def MAX_CONNECTIONS(self, value: int) -> None:
        # Allow tests to override via instance attribute
        object.__setattr__(self, "_max_connections_override", value)

    @property
    def MAX_CONNECTIONS_PER_IP(self) -> int:
        override = getattr(self, "_max_connections_per_ip_override", None)
        if override is not None:
            return int(override)
        return int(_sse_config_value("maxConnectionsPerIP", self.DEFAULT_MAX_CONNECTIONS_PER_IP))

    @MAX_CONNECTIONS_PER_IP.setter
    def MAX_CONNECTIONS_PER_IP(self, value: int) -> None:
        object.__setattr__(self, "_max_connections_per_ip_override", value)

    def _effective_max_connections(self) -> int:
        override = getattr(self, "_max_connections_override", None)
        if override is not None:
            return int(override)
        return int(_sse_config_value("maxTotalConnections", self.DEFAULT_MAX_CONNECTIONS))

    async def connect(
        self,
        connection_id: str | None = None,
        last_event_id: str | None = None,
        client_ip: str | None = None,
        api_key: str | None = None,
    ) -> SSEConnection:
        """
        Establish new SSE connection

        :param connection_id: Connection ID (auto-generate if None)
        :param last_event_id: Last received event ID (for reconnection)
        :param client_ip: Client IP address
        :param api_key: API key for authentication
        :return: SSE connection instance
        :raises InternalException: If connection limit is reached
        """
        if not connection_id:
            connection_id = uuid.new_uuid()

        queue: asyncio.Queue[ISSEEvent] = asyncio.Queue(maxsize=SSEConnection.max_queue_size())

        context = Context(context_id=connection_id)
        context.with_value("client_ip", client_ip)
        context.with_value("api_key", api_key)
        context.with_value("last_event_id", last_event_id)

        connection = SSEConnection(
            connection_id=connection_id,
            queue=queue,
            context=context,
            last_event_id=last_event_id,
        )

        with self._lock:
            max_connections = self._effective_max_connections()
            if len(self._connections) >= max_connections:
                raise InternalException(f"Maximum connections reached: {max_connections}")

            if client_ip:
                ip_count = sum(
                    1 for conn in self._connections.values() if conn.context.context.get("client_ip") == client_ip
                )
                if ip_count >= self.MAX_CONNECTIONS_PER_IP:
                    raise InternalException(f"Maximum connections per IP reached: {self.MAX_CONNECTIONS_PER_IP}")

            # Replace duplicate id by closing old entry first
            old = self._connections.get(connection_id)
            if old is not None:
                old._closed = True
            self._connections[connection_id] = connection
            total = len(self._connections)

        log.request_id(connection_id).info(
            f"SSE connection established: {connection_id}, "
            f"last_event_id={last_event_id}, "
            f"client_ip={client_ip}, "
            f"total_connections={total}"
        )

        return connection

    async def disconnect(self, connection_id: str) -> None:
        """
        Disconnect and remove connection

        :param connection_id: Connection ID
        """
        connection = self.get_connection(connection_id)
        if connection:
            await connection.disconnect()
            with self._lock:
                self._connections.pop(connection_id, None)
                remaining = len(self._connections)

            log.request_id(connection_id).info(
                f"SSE connection removed: {connection_id}, remaining_connections={remaining}"
            )

    def get_connection(self, connection_id: str) -> SSEConnection | None:
        """
        Get connection by ID

        :param connection_id: Connection ID
        :return: Connection instance or None
        """
        with self._lock:
            return self._connections.get(connection_id)

    def get_all_connections(self) -> list[ISSEConnection]:
        """
        Get all active connections

        :return: List of connections
        """
        with self._lock:
            return list(self._connections.values())

    async def broadcast(self, event: ISSEEvent) -> int:
        """
        Broadcast event to all active connections

        :param event: SSE event
        :return: Number of connections that received the event
        """
        connections = self.get_all_connections()
        active_connections = [conn for conn in connections if not conn.is_closed()]

        results = await asyncio.gather(
            *[conn.send(event) for conn in active_connections],
            return_exceptions=True,
        )

        success_count = sum(1 for result in results if result is True)

        log.info(f"SSE broadcast: event={event.event}, connections={len(active_connections)}, success={success_count}")

        return success_count

    def get_stats(self) -> dict:
        """
        Get global connection statistics

        :return: Statistics dict
        """
        connections = [c for c in self.get_all_connections() if isinstance(c, SSEConnection)]
        return {
            "total_connections": len(connections),
            "active_connections": len([c for c in connections if not c.is_closed()]),
            "connection_stats": [conn.get_stats() for conn in connections],
            "total_dropped_events": sum(conn._dropped_events for conn in connections),
        }

DEFAULT_MAX_CONNECTIONS class-attribute instance-attribute 

DEFAULT_MAX_CONNECTIONS = 100

DEFAULT_MAX_CONNECTIONS_PER_IP class-attribute instance-attribute 

DEFAULT_MAX_CONNECTIONS_PER_IP = 5

MAX_CONNECTIONS property writable 

MAX_CONNECTIONS: int

MAX_CONNECTIONS_PER_IP property writable 

MAX_CONNECTIONS_PER_IP: int

connect async 

connect(
    connection_id: str | None = None,
    last_event_id: str | None = None,
    client_ip: str | None = None,
    api_key: str | None = None,
) -> SSEConnection

Establish new SSE connection

Parameters:

Name Type Description Default
connection_id str | None

Connection ID (auto-generate if None)

 None
last_event_id str | None

Last received event ID (for reconnection)

 None
client_ip str | None

Client IP address

 None
api_key str | None

API key for authentication

 None

Returns:

Type Description
SSEConnection

SSE connection instance

Raises:

Type Description
InternalException

If connection limit is reached
Source code in pypepper/network/http/sse/connection.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
async def connect(
    self,
    connection_id: str | None = None,
    last_event_id: str | None = None,
    client_ip: str | None = None,
    api_key: str | None = None,
) -> SSEConnection:
    """
    Establish new SSE connection

    :param connection_id: Connection ID (auto-generate if None)
    :param last_event_id: Last received event ID (for reconnection)
    :param client_ip: Client IP address
    :param api_key: API key for authentication
    :return: SSE connection instance
    :raises InternalException: If connection limit is reached
    """
    if not connection_id:
        connection_id = uuid.new_uuid()

    queue: asyncio.Queue[ISSEEvent] = asyncio.Queue(maxsize=SSEConnection.max_queue_size())

    context = Context(context_id=connection_id)
    context.with_value("client_ip", client_ip)
    context.with_value("api_key", api_key)
    context.with_value("last_event_id", last_event_id)

    connection = SSEConnection(
        connection_id=connection_id,
        queue=queue,
        context=context,
        last_event_id=last_event_id,
    )

    with self._lock:
        max_connections = self._effective_max_connections()
        if len(self._connections) >= max_connections:
            raise InternalException(f"Maximum connections reached: {max_connections}")

        if client_ip:
            ip_count = sum(
                1 for conn in self._connections.values() if conn.context.context.get("client_ip") == client_ip
            )
            if ip_count >= self.MAX_CONNECTIONS_PER_IP:
                raise InternalException(f"Maximum connections per IP reached: {self.MAX_CONNECTIONS_PER_IP}")

        # Replace duplicate id by closing old entry first
        old = self._connections.get(connection_id)
        if old is not None:
            old._closed = True
        self._connections[connection_id] = connection
        total = len(self._connections)

    log.request_id(connection_id).info(
        f"SSE connection established: {connection_id}, "
        f"last_event_id={last_event_id}, "
        f"client_ip={client_ip}, "
        f"total_connections={total}"
    )

    return connection

disconnect async 

disconnect(connection_id: str) -> None

Disconnect and remove connection

:param connection_id: Connection ID

Source code in pypepper/network/http/sse/connection.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
async def disconnect(self, connection_id: str) -> None:
    """
    Disconnect and remove connection

    :param connection_id: Connection ID
    """
    connection = self.get_connection(connection_id)
    if connection:
        await connection.disconnect()
        with self._lock:
            self._connections.pop(connection_id, None)
            remaining = len(self._connections)

        log.request_id(connection_id).info(
            f"SSE connection removed: {connection_id}, remaining_connections={remaining}"
        )

get_connection 

get_connection(connection_id: str) -> SSEConnection | None

Get connection by ID

Parameters:

Name Type Description Default
connection_id str

Connection ID

 required

Returns:

Type Description
SSEConnection | None

Connection instance or None
Source code in pypepper/network/http/sse/connection.py 
239
240
241
242
243
244
245
246
247
def get_connection(self, connection_id: str) -> SSEConnection | None:
    """
    Get connection by ID

    :param connection_id: Connection ID
    :return: Connection instance or None
    """
    with self._lock:
        return self._connections.get(connection_id)

get_all_connections 

get_all_connections() -> list[ISSEConnection]

Get all active connections

:return: List of connections

Source code in pypepper/network/http/sse/connection.py 
249
250
251
252
253
254
255
256
def get_all_connections(self) -> list[ISSEConnection]:
    """
    Get all active connections

    :return: List of connections
    """
    with self._lock:
        return list(self._connections.values())

broadcast async 

broadcast(event: ISSEEvent) -> int

Broadcast event to all active connections

Parameters:

Name Type Description Default
event ISSEEvent

SSE event

 required

Returns:

Type Description
int

Number of connections that received the event
Source code in pypepper/network/http/sse/connection.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
async def broadcast(self, event: ISSEEvent) -> int:
    """
    Broadcast event to all active connections

    :param event: SSE event
    :return: Number of connections that received the event
    """
    connections = self.get_all_connections()
    active_connections = [conn for conn in connections if not conn.is_closed()]

    results = await asyncio.gather(
        *[conn.send(event) for conn in active_connections],
        return_exceptions=True,
    )

    success_count = sum(1 for result in results if result is True)

    log.info(f"SSE broadcast: event={event.event}, connections={len(active_connections)}, success={success_count}")

    return success_count

get_stats 

get_stats() -> dict

Get global connection statistics

:return: Statistics dict

Source code in pypepper/network/http/sse/connection.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
def get_stats(self) -> dict:
    """
    Get global connection statistics

    :return: Statistics dict
    """
    connections = [c for c in self.get_all_connections() if isinstance(c, SSEConnection)]
    return {
        "total_connections": len(connections),
        "active_connections": len([c for c in connections if not c.is_closed()]),
        "connection_stats": [conn.get_stats() for conn in connections],
        "total_dropped_events": sum(conn._dropped_events for conn in connections),
    }

pypepper.network.http.sse.stream

SSE streaming response helpers.

sse_stream async 

sse_stream(
    request: Request,
    handler: ISSEHandler,
    connection_id: str | None = None,
) -> AsyncIterable[str]

SSE stream generator

Yields SSE-formatted strings ready for transmission.

Parameters:

Name Type Description Default
request Request

FastAPI Request object

 required
handler ISSEHandler

SSE handler

 required
connection_id str | None

Connection ID (optional, auto-generate if None)

 None

Returns:

Type Description
AsyncIterable[str]

AsyncIterable of SSE-formatted strings
Source code in pypepper/network/http/sse/stream.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
async def sse_stream(
    request: Request,
    handler: ISSEHandler,
    connection_id: str | None = None,
) -> AsyncIterable[str]:
    """
    SSE stream generator

    Yields SSE-formatted strings ready for transmission.

    :param request: FastAPI Request object
    :param handler: SSE handler
    :param connection_id: Connection ID (optional, auto-generate if None)
    :return: AsyncIterable of SSE-formatted strings
    """
    # Extract metadata from request
    last_event_id = request.headers.get("Last-Event-ID")
    client_host = request.client.host if request.client is not None else None
    client_ip = getattr(request.state, "client_ip", client_host)
    api_key = getattr(request.state, "api_key", None)

    # Establish connection
    connection = await connection_manager.connect(
        connection_id=connection_id,
        last_event_id=last_event_id,
        client_ip=client_ip,
        api_key=api_key,
    )

    try:
        # Trigger on_connect callback
        await handler.on_connect(connection)

        # Create event generation task
        event_gen_task = asyncio.create_task(_consume_handler_events(handler, connection))

        # Stream events from queue
        while not await request.is_disconnected():
            try:
                # Wait for event with timeout (for heartbeat)
                event = await asyncio.wait_for(
                    connection._queue.get(),
                    timeout=_stream_timeout_seconds(),
                )

                # Convert to ServerSentEvent and serialize to SSE format
                sse_event = event.to_server_sent_event()
                yield _serialize_sse_event(sse_event)

            except asyncio.TimeoutError:
                # Timeout: send heartbeat
                from pypepper.network.http.sse.event import SSEEvent

                sse_event = SSEEvent.ping().to_server_sent_event()
                yield _serialize_sse_event(sse_event)
                continue

            except Exception as e:
                log.error(f"SSE stream error: {e}")
                break

        # Cancel event generation task
        event_gen_task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await event_gen_task

    finally:
        # Trigger on_disconnect callback
        await handler.on_disconnect(connection)

        # Cleanup connection
        await connection_manager.disconnect(connection.connection_id)

pypepper.network.http.sse.handlers

SSE route handlers.

BaseSSEHandler

Bases: ISSEHandler

Base SSE Handler (default implementation)

Source code in pypepper/network/http/sse/handlers.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class BaseSSEHandler(ISSEHandler):
    """Base SSE Handler (default implementation)"""

    async def on_connect(self, connection: ISSEConnection) -> None:
        """
        Called when client connects

        :param connection: SSE connection
        """
        # Send welcome message
        await connection.send(
            SSEEvent(
                data={
                    "message": "Connected",
                    "connection_id": connection.connection_id,
                },
                event="connect",
                id=connection.connection_id,
            )
        )

    async def on_disconnect(self, connection: ISSEConnection) -> None:
        """
        Called when client disconnects

        :param connection: SSE connection
        """
        # Default: no action on disconnect
        pass

    async def generate_events(
        self,
        connection: ISSEConnection,
    ) -> AsyncIterator[ISSEEvent]:
        """
        Generate events for the connection (default: no events)

        :param connection: SSE connection
        :return: Async iterator of SSE events
        """
        # Default: no events generated
        # Subclasses should override this method
        return
        yield  # Make it a generator

on_connect async 

on_connect(connection: ISSEConnection) -> None

Called when client connects

:param connection: SSE connection

Source code in pypepper/network/http/sse/handlers.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
async def on_connect(self, connection: ISSEConnection) -> None:
    """
    Called when client connects

    :param connection: SSE connection
    """
    # Send welcome message
    await connection.send(
        SSEEvent(
            data={
                "message": "Connected",
                "connection_id": connection.connection_id,
            },
            event="connect",
            id=connection.connection_id,
        )
    )

on_disconnect async 

on_disconnect(connection: ISSEConnection) -> None

Called when client disconnects

:param connection: SSE connection

Source code in pypepper/network/http/sse/handlers.py 
30
31
32
33
34
35
36
37
async def on_disconnect(self, connection: ISSEConnection) -> None:
    """
    Called when client disconnects

    :param connection: SSE connection
    """
    # Default: no action on disconnect
    pass

generate_events async 

generate_events(
    connection: ISSEConnection,
) -> AsyncIterator[ISSEEvent]

Generate events for the connection (default: no events)

Parameters:

Name Type Description Default
connection ISSEConnection

SSE connection

 required

Returns:

Type Description
AsyncIterator[ISSEEvent]

Async iterator of SSE events
Source code in pypepper/network/http/sse/handlers.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
async def generate_events(
    self,
    connection: ISSEConnection,
) -> AsyncIterator[ISSEEvent]:
    """
    Generate events for the connection (default: no events)

    :param connection: SSE connection
    :return: Async iterator of SSE events
    """
    # Default: no events generated
    # Subclasses should override this method
    return
    yield  # Make it a generator

EchoSSEHandler

Bases: BaseSSEHandler

Echo SSE Handler (for testing)

Source code in pypepper/network/http/sse/handlers.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class EchoSSEHandler(BaseSSEHandler):
    """Echo SSE Handler (for testing)"""

    def __init__(self, messages: list[str] | None = None, interval: float = 1.0):
        """
        Initialize Echo handler

        :param messages: List of messages to echo
        :param interval: Interval between messages (seconds)
        """
        self.messages = messages or ["Hello", "World", "SSE", "Test"]
        self.interval = interval

    async def generate_events(
        self,
        connection: ISSEConnection,
    ) -> AsyncIterator[ISSEEvent]:
        """
        Generate echo events

        :param connection: SSE connection
        :return: Async iterator of SSE events
        """
        import asyncio

        for i, msg in enumerate(self.messages):
            await asyncio.sleep(self.interval)
            yield SSEEvent(
                data={"message": msg, "index": i},
                event="echo",
                id=f"echo-{i}",
            )

messages instance-attribute 

messages = messages or ['Hello', 'World', 'SSE', 'Test']

interval instance-attribute 

interval = interval

generate_events async 

generate_events(
    connection: ISSEConnection,
) -> AsyncIterator[ISSEEvent]

Generate echo events

Parameters:

Name Type Description Default
connection ISSEConnection

SSE connection

 required

Returns:

Type Description
AsyncIterator[ISSEEvent]

Async iterator of SSE events
Source code in pypepper/network/http/sse/handlers.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def generate_events(
    self,
    connection: ISSEConnection,
) -> AsyncIterator[ISSEEvent]:
    """
    Generate echo events

    :param connection: SSE connection
    :return: Async iterator of SSE events
    """
    import asyncio

    for i, msg in enumerate(self.messages):
        await asyncio.sleep(self.interval)
        yield SSEEvent(
            data={"message": msg, "index": i},
            event="echo",
            id=f"echo-{i}",
        )

CounterSSEHandler

Bases: BaseSSEHandler

Counter SSE Handler (counting example)

Source code in pypepper/network/http/sse/handlers.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class CounterSSEHandler(BaseSSEHandler):
    """Counter SSE Handler (counting example)"""

    def __init__(self, start: int = 0, end: int = 10, interval: float = 1.0):
        """
        Initialize Counter handler

        :param start: Start value
        :param end: End value
        :param interval: Interval between counts (seconds)
        """
        self.start = start
        self.end = end
        self.interval = interval

    async def generate_events(
        self,
        connection: ISSEConnection,
    ) -> AsyncIterator[ISSEEvent]:
        """
        Generate counter events

        :param connection: SSE connection
        :return: Async iterator of SSE events
        """
        import asyncio

        for count in range(self.start, self.end + 1):
            await asyncio.sleep(self.interval)
            yield SSEEvent(
                data={"count": count},
                event="counter",
                id=f"count-{count}",
            )

start instance-attribute 

start = start

end instance-attribute 

end = end

interval instance-attribute 

interval = interval

generate_events async 

generate_events(
    connection: ISSEConnection,
) -> AsyncIterator[ISSEEvent]

Generate counter events

Parameters:

Name Type Description Default
connection ISSEConnection

SSE connection

 required

Returns:

Type Description
AsyncIterator[ISSEEvent]

Async iterator of SSE events
Source code in pypepper/network/http/sse/handlers.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
async def generate_events(
    self,
    connection: ISSEConnection,
) -> AsyncIterator[ISSEEvent]:
    """
    Generate counter events

    :param connection: SSE connection
    :return: Async iterator of SSE events
    """
    import asyncio

    for count in range(self.start, self.end + 1):
        await asyncio.sleep(self.interval)
        yield SSEEvent(
            data={"count": count},
            event="counter",
            id=f"count-{count}",
        )

pypepper.network.http.sse.security

SSE API key authentication helpers.

AUTH_OFF_BLOCKED_DETAIL module-attribute 

AUTH_OFF_BLOCKED_DETAIL = f"SSE authentication cannot be disabled without {_AUTH_OFF_ENV}=1 (or true/yes/on); set sse.authentication.enabled: true and inject validKeys, or export {_AUTH_OFF_ENV}=1 for local experiments only"

sse_security module-attribute 

sse_security = SSESecurityManager()

SSESecurityManager

SSE Security Manager (API Key authentication + Rate limiting).

Source code in pypepper/network/http/sse/security.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class SSESecurityManager:
    """SSE Security Manager (API Key authentication + Rate limiting)."""

    _instance: SSESecurityManager | None = None
    _init_lock = Lock()
    _rate_limit_cache: Cache
    _rate_limit_lock: Lock

    def __new__(cls) -> SSESecurityManager:
        with cls._init_lock:
            if cls._instance is None:
                inst = super().__new__(cls)
                inst._rate_limit_cache = Cache(maxsize=1000, ttl=60)
                inst._rate_limit_lock = Lock()
                cls._instance = inst
            return cls._instance

    def __init__(self) -> None:
        pass

    def _validate_api_key(self, api_key: str | None) -> bool:
        """Validate API Key; returns True if valid."""
        if not api_key:
            return False

        sse_config = config.get_yml_config().sse
        if not sse_config.authentication.enabled:
            if not _auth_off_allowed():
                _warn_auth_off_blocked_once()
                return False
            _warn_auth_disabled_once()
            return True

        valid_keys = list(sse_config.authentication.validKeys or [])
        return any(compare_digest(api_key, key) for key in valid_keys if isinstance(key, str))

    def _check_rate_limit(self, client_id: str) -> bool:
        """Check rate limit (simple counter); returns True if within limit."""
        sse_config = config.get_yml_config().sse

        if not sse_config.rateLimit.enabled:
            return True

        max_requests = sse_config.rateLimit.maxRequestsPerMinute
        key = f"rate_limit:{client_id}"

        with self._rate_limit_lock:
            count = self._rate_limit_cache.get(key) or 0
            if count >= max_requests:
                return False
            self._rate_limit_cache.set(key, count + 1)
            return True

    @staticmethod
    def validate_api_key(api_key: str | None) -> bool:
        """Validate API Key (delegates to the process singleton)."""
        return sse_security._validate_api_key(api_key)

    @staticmethod
    def check_rate_limit(client_id: str) -> bool:
        """Check rate limit (delegates to the process singleton)."""
        return sse_security._check_rate_limit(client_id)

validate_api_key staticmethod 

validate_api_key(api_key: str | None) -> bool

Validate API Key (delegates to the process singleton).

Source code in pypepper/network/http/sse/security.py 
130
131
132
133
@staticmethod
def validate_api_key(api_key: str | None) -> bool:
    """Validate API Key (delegates to the process singleton)."""
    return sse_security._validate_api_key(api_key)

check_rate_limit staticmethod 

check_rate_limit(client_id: str) -> bool

Check rate limit (delegates to the process singleton).

Source code in pypepper/network/http/sse/security.py 
135
136
137
138
@staticmethod
def check_rate_limit(client_id: str) -> bool:
    """Check rate limit (delegates to the process singleton)."""
    return sse_security._check_rate_limit(client_id)

reset_auth_disabled_warning 

reset_auth_disabled_warning() -> None

Reset one-shot auth-off warnings (tests).

Source code in pypepper/network/http/sse/security.py 
69
70
71
72
73
74
def reset_auth_disabled_warning() -> None:
    """Reset one-shot auth-off warnings (tests)."""
    global _auth_disabled_warned, _auth_off_blocked_warned
    with _auth_warn_lock:
        _auth_disabled_warned = False
        _auth_off_blocked_warned = False

require_sse_api_key 

require_sse_api_key(func: Callable) -> Callable

API Key authentication decorator

Supports: 1. Header: X-API-Key: your-key 2. Header: Authorization: Bearer your-key

Query-string API keys are rejected to avoid log/Referer leakage.

When sse.authentication.enabled is false, requests are rejected with 503 unless PYPEPPER_SSE_ALLOW_AUTH_OFF is set (local experiments only).

Usage: @app.get('/sse') @require_sse_api_key async def sse_endpoint(request: Request): ...

Source code in pypepper/network/http/sse/security.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def require_sse_api_key(func: Callable) -> Callable:
    """
    API Key authentication decorator

    Supports:
    1. Header: X-API-Key: your-key
    2. Header: Authorization: Bearer your-key

    Query-string API keys are rejected to avoid log/Referer leakage.

    When ``sse.authentication.enabled`` is false, requests are rejected with 503
    unless ``PYPEPPER_SSE_ALLOW_AUTH_OFF`` is set (local experiments only).

    Usage:
        @app.get('/sse')
        @require_sse_api_key
        async def sse_endpoint(request: Request):
            ...
    """

    @wraps(func)
    async def wrapper(request: Request, *args: Any, **kwargs: Any):
        sse_config = config.get_yml_config().sse
        if not sse_config.authentication.enabled and not _auth_off_allowed():
            _warn_auth_off_blocked_once()
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail=AUTH_OFF_BLOCKED_DETAIL,
            )

        authorization = request.headers.get("Authorization", "")
        bearer = ""
        if authorization.lower().startswith("bearer "):
            bearer = authorization[7:].strip()

        api_key = request.headers.get("X-API-Key") or bearer or None

        # Validate API Key
        if not SSESecurityManager.validate_api_key(api_key):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid or missing API key",
                headers={"WWW-Authenticate": "Bearer"},
            )

        # Rate limit check
        client_id = api_key or ""
        if not SSESecurityManager.check_rate_limit(client_id):
            max_requests = sse_config.rateLimit.maxRequestsPerMinute
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail=f"Rate limit exceeded (max {max_requests} requests/minute)",
            )

        # Store metadata in request state for later use
        request.state.api_key = api_key
        request.state.client_ip = request.client.host if request.client is not None else None

        return await func(request, *args, **kwargs)

    return wrapper

pypepper.network.http.sse.event

SSE event payload helpers.

SSEEvent

Bases: ISSEEvent

SSE Event wrapper Independent event definition for FastAPI ServerSentEvent

Source code in pypepper/network/http/sse/event.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class SSEEvent(ISSEEvent):
    """
    SSE Event wrapper
    Independent event definition for FastAPI ServerSentEvent
    """

    def __init__(
        self,
        data: Any = None,
        raw_data: str | None = None,
        event: str | None = None,
        id: str | None = None,
        retry: int | None = None,
        comment: str | None = None,
    ):
        """
        Initialize SSE event

        :param data: JSON-serializable data (mutually exclusive with raw_data)
        :param raw_data: Raw string data (mutually exclusive with data)
        :param event: Event type name
        :param id: Event ID (for reconnection)
        :param retry: Reconnection interval in milliseconds
        :param comment: Comment line (for keep-alive heartbeat)
        """
        self.data = data
        self.raw_data = raw_data
        self.event = event
        self.id = id
        self.retry = retry
        self.comment = comment

    def to_server_sent_event(self) -> ServerSentEvent:
        """
        Convert to FastAPI ServerSentEvent
        :return: ServerSentEvent instance
        """
        return ServerSentEvent(
            data=self.data,
            raw_data=self.raw_data,
            event=self.event,
            id=self.id,
            retry=self.retry,
            comment=self.comment,
        )

    @staticmethod
    def ping(comment: str = "heartbeat") -> SSEEvent:
        """
        Create heartbeat event
        :param comment: Comment text
        :return: SSEEvent instance
        """
        return SSEEvent(comment=comment)

    @staticmethod
    def reconnect(retry_ms: int = 3000) -> SSEEvent:
        """
        Create reconnection configuration event
        :param retry_ms: Retry interval in milliseconds
        :return: SSEEvent instance
        """
        return SSEEvent(retry=retry_ms)

data instance-attribute 

data = data

raw_data instance-attribute 

raw_data = raw_data

event instance-attribute 

event = event

id instance-attribute 

id = id

retry instance-attribute 

retry = retry

comment instance-attribute 

comment = comment

to_server_sent_event 

to_server_sent_event() -> ServerSentEvent

Convert to FastAPI ServerSentEvent :return: ServerSentEvent instance

Source code in pypepper/network/http/sse/event.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
def to_server_sent_event(self) -> ServerSentEvent:
    """
    Convert to FastAPI ServerSentEvent
    :return: ServerSentEvent instance
    """
    return ServerSentEvent(
        data=self.data,
        raw_data=self.raw_data,
        event=self.event,
        id=self.id,
        retry=self.retry,
        comment=self.comment,
    )

ping staticmethod 

ping(comment: str = 'heartbeat') -> SSEEvent

Create heartbeat event

Parameters:

Name Type Description Default
comment str

Comment text

 'heartbeat'

Returns:

Type Description
SSEEvent

SSEEvent instance
Source code in pypepper/network/http/sse/event.py 
58
59
60
61
62
63
64
65
@staticmethod
def ping(comment: str = "heartbeat") -> SSEEvent:
    """
    Create heartbeat event
    :param comment: Comment text
    :return: SSEEvent instance
    """
    return SSEEvent(comment=comment)

reconnect staticmethod 

reconnect(retry_ms: int = 3000) -> SSEEvent

Create reconnection configuration event

Parameters:

Name Type Description Default
retry_ms int

Retry interval in milliseconds

 3000

Returns:

Type Description
SSEEvent

SSEEvent instance
Source code in pypepper/network/http/sse/event.py 
67
68
69
70
71
72
73
74
@staticmethod
def reconnect(retry_ms: int = 3000) -> SSEEvent:
    """
    Create reconnection configuration event
    :param retry_ms: Retry interval in milliseconds
    :return: SSEEvent instance
    """
    return SSEEvent(retry=retry_ms)

new_sse_event 

new_sse_event(
    data: Any = None,
    event: str | None = None,
    id: str | None = None,
) -> SSEEvent

Create SSE event

Parameters:

Name Type Description Default
data Any

Event data

 None
event str | None

Event type

 None
id str | None

Event ID

 None

Returns:

Type Description
SSEEvent

SSEEvent instance
Source code in pypepper/network/http/sse/event.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def new_sse_event(
    data: Any = None,
    event: str | None = None,
    id: str | None = None,
) -> SSEEvent:
    """
    Create SSE event

    :param data: Event data
    :param event: Event type
    :param id: Event ID
    :return: SSEEvent instance
    """
    return SSEEvent(data=data, event=event, id=id)