Python SDK - Stream Observer

Pass a StreamObserver via NexusConfig.stream_observer to receive structured callbacks for every SSE lifecycle transition. The observer is the canonical way to wire SDK state into ops dashboards, custom telemetry, or load-test tooling - far more reliable than scraping NexusLogger output by substring match. The observer is opt-in. When NexusConfig.stream_observer is None the SDK uses NOOP_STREAM_OBSERVER at zero overhead.

The interface

from datetime import datetime
from typing import Optional, Protocol, runtime_checkable

@runtime_checkable
class StreamObserver(Protocol):
    def on_connected(self) -> None: ...
    def on_disconnected(self, cause: Optional[BaseException]) -> None: ...
    def on_event(self, name: str) -> None: ...
    def on_reconnect_attempt(self, attempt: int) -> None: ...
    def on_fallback(self, reason: str) -> None: ...
    def on_quarantined(self, reason: str, expires_at: datetime) -> None: ...  # v0.4.0

The protocol is structural - any object with the matching method signatures qualifies. Methods are looked up via getattr, so consumers may omit any method they don’t care about (the SDK handles missing-attribute cases gracefully).

Hook semantics

Hook	Fires when	Argument
`on_connected()`	`GET /v1/stream` returns `200 OK` and the SDK is ready to read events	none
`on_disconnected(cause)`	An active stream connection ends - clean disconnect, transport error, or server-initiated close	`None` for clean shutdown, the underlying exception otherwise
`on_event(name)`	A whitelisted SSE event arrives, before the SDK runs the event-triggered sync	event name string (e.g. `"config.updated"`)
`on_reconnect_attempt(n)`	The SDK is about to back off and retry after a transport error	1-based attempt counter (max 3 with current threshold)
`on_fallback(reason)`	The SDK enters a degraded state	`FALLBACK_REASON_MAX_ERRORS` or `FALLBACK_REASON_RATE_LIMITED`
`on_quarantined(reason, expires_at)` (v0.4.0)	Server returned `429` with a quarantine body on either the sync or stream path	`reason` string; `expires_at` timezone-aware `datetime`

`on_fallback` reasons

from westyx_nexus import (
    FALLBACK_REASON_MAX_ERRORS,
    FALLBACK_REASON_RATE_LIMITED,
)

Constant	When	Stream after
`'max-errors'`	3 consecutive transport errors - backoff exhausted, the SDK gives up	TTL polling only; reconnect requires `connect_stream()`
`'rate-limited'`	server returned `429 Too Many Requests` (tier connection-limit hit)	depends on the response - see below

rate-limited semantics (v0.3.1+). This reason now fires in both of these situations:

Server included a parseable Retry-After header. The SDK fires on_fallback(FALLBACK_REASON_RATE_LIMITED), sleeps the indicated delay clamped to [5, 300] seconds, and then reconnects automatically. The 429 does not count against _MAX_STREAM_ERRORS. Expect a matching on_connected() shortly after the sleep window.
No parseable Retry-After. Legacy behaviour: the SDK fires on_fallback(FALLBACK_REASON_RATE_LIMITED), closes the stream, and stays on TTL polling. No automatic reconnect - connect_stream() is required to recover.

Threading contract

For the synchronous NexusClient, observer methods are called synchronously from the SDK’s stream thread. For the async AsyncNexusClient, observer methods are called synchronously from the asyncio task running the stream - they are NOT awaited even if you define them as coroutines.

In both cases observer methods MUST return promptly. The SDK does not protect against blocking observers.

If your observer does anything substantial (network I/O, lock contention, disk writes), copy the argument and dispatch the work to your own thread / task before returning:

import queue, threading
class SlowObserver:
    def __init__(self) -> None:
        self.q: queue.Queue[str] = queue.Queue(maxsize=1024)
        threading.Thread(target=self._drain, daemon=True).start()

    def on_event(self, name: str) -> None:
        try:
            self.q.put_nowait(name)
        except queue.Full:
            pass  # Drop on overflow - backpressure on the SDK would stall reads.

    def _drain(self) -> None:
        while True:
            event = self.q.get()
            ... # process(event)

Counter pattern

from dataclasses import dataclass
from threading import Lock
from westyx_nexus import NexusClient, NexusConfig

@dataclass
class CounterObserver:
    events: int = 0
    reconnects: int = 0
    fallbacks: int = 0
    _lock: Lock = Lock()

    def on_event(self, name: str) -> None:
        with self._lock:
            self.events += 1

    def on_reconnect_attempt(self, attempt: int) -> None:
        with self._lock:
            self.reconnects += 1

    def on_fallback(self, reason: str) -> None:
        with self._lock:
            self.fallbacks += 1

obs = CounterObserver()
client = NexusClient.create(NexusConfig(
    base_url="https://...",
    api_key="wxs_...",
    stream_observer=obs,
))

Active-connection gauge pattern

class GaugeObserver:
    def __init__(self) -> None:
        self.connected = False

    def on_connected(self) -> None:
        self.connected = True

    def on_disconnected(self, cause) -> None:
        self.connected = False

    def on_fallback(self, reason: str) -> None:
        self.connected = False

on_fallback is included because after the 3-strike FALLBACK_REASON_MAX_ERRORS case the SDK will not reconnect - the gauge should stay False until you call connect_stream() again. For FALLBACK_REASON_RATE_LIMITED with a Retry-After (v0.3.1+) the SDK does reconnect on its own; the gauge will flip back to True on the next on_connected().

Overview

Backend SDKs

Browser SDKs

OpenFeature Providers

Python SDK - Stream Observer

The interface

Hook semantics

`on_fallback` reasons

Threading contract

Counter pattern

Active-connection gauge pattern

​The interface

​Hook semantics

​on_fallback reasons

​Threading contract

​Counter pattern

​Active-connection gauge pattern

The interface

Hook semantics

`on_fallback` reasons

Threading contract

Counter pattern

Active-connection gauge pattern