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)

If your observer raises

Changed in v0.11.0. An exception from an observer method is caught and absorbed - a faulty observer can never break the stream, stall a sync, or propagate into your own code paths. It is also logged. The SDK reports it through NexusLogger at error level, naming the hook that failed and attaching the original exception as the cause:

nexus: stream observer on_connected() raised - ignoring: division by zero

Before v0.11.0 the exception was discarded with no trace anywhere, so an observer that threw on every event was indistinguishable from one that was never called. If you set no logger the fault is still absorbed - it just goes unreported, as before.

This follows the OpenTelemetry error-handling specification, which covers the same situation from both sides: “API methods that accept external callbacks MUST handle all errors”, and “whenever the library suppresses an error that would otherwise have been exposed to the user, the library SHOULD log the error”. A missing hook is not a fault - the observer protocol is partial by design. Implement only the callbacks you need; the rest are skipped silently.

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

If your observer raises

Counter pattern

Active-connection gauge pattern

​The interface

​Hook semantics

​on_fallback reasons

​Threading contract

​If your observer raises

​Counter pattern

​Active-connection gauge pattern

The interface

Hook semantics

`on_fallback` reasons

Threading contract

If your observer raises

Counter pattern

Active-connection gauge pattern