API reference

The public API is exported from the top-level fastapi_stream_lease package. Internal modules, names prefixed with _, and Lua script layouts are implementation details.

Manager

StreamLeaseManager

StreamLeaseManager(
    redis: Any,
    config: LeaseConfig | None = None,
    telemetry: TelemetryAdapter | None = None,
    metrics: Any = None,
)

Coordinates distributed stream concurrency leases backed by atomic Redis Lua scripts.

is_closing property

is_closing: bool

Whether the manager is currently in the process of shutting down.

active_leases_count property

active_leases_count: int

Count of active stream leases locally tracked in this manager process.

active_leases property

active_leases: frozenset[StreamLease]

Snapshot of active stream leases currently tracked by this manager.

circuit_state property

circuit_state: CircuitState | None

Current operational state of the circuit breaker, or None if disabled.

metrics property writable

metrics: Any

Backwards compatibility property alias for self.telemetry.

drain async

drain(timeout: float = 5.0) -> None

Wait for all queued lifecycle callbacks to complete execution.

close async

close(drain: bool = True, timeout: float = 5.0) -> None

Shut down the stream lease manager, draining active leases and hooks.

When drain=True, rejects new acquisitions and waits up to timeout seconds for active streams to complete naturally. If the timeout expires before all leases finish, any remaining leases are force-released in Redis to avoid leaving orphan reservations.

acquire async

acquire(
    user_id: str | int, timeout: float | None = None
) -> StreamLease

Acquire a new stream lease for the given user.

Parameters:

Name	Type	Description	Default
user_id	str | int	Unique identifier for the user/principal.	required
timeout	float | None	Optional maximum duration (in seconds) to wait for an available slot if limits are currently reached. If None, defaults to config.acquire_timeout. Set to 0 for strict immediate fail-fast.	None

Raises:

Type	Description
StreamLeaseUnavailable	If the manager is shutting down.
StreamLeaseRejected	If user or global concurrency limits are exceeded.
ValueError	If timeout is negative or non-finite.
RuntimeError	If Redis evaluation fails unexpectedly.

renew async

renew(lease: StreamLease) -> bool

Renew an active lease, extending its TTL in Redis.

Returns:

Name	Type	Description
bool	bool	True if successfully extended, False if the lease expired or was evicted in Redis.

Raises: StreamLeaseUnavailable: If the Redis backend is unreachable.

release async

release(lease: StreamLease) -> None

Release an active lease immediately from Redis.

get_active_count async

get_active_count(user_id: str | int | None = None) -> int

Return the current number of active (non-expired) streams for a user or globally.

lease async

lease(
    user_id: str | int,
    renew_interval: float | None = None,
    timeout: float | None = None,
) -> AsyncIterator[StreamLease]

Context manager for acquiring and safely releasing a stream lease for scoped executions (such as WebSockets, background tasks, or pub/sub loops).

For HTTP StreamingResponse (SSE / LLM tokens), use await manager.stream(...) or return lease.as_streaming_response(...) instead.

Example

async with lease_manager.lease(user_id=42) as lease: while True: msg = await websocket.receive_text() ...

register_exception_handlers

register_exception_handlers(app: Any) -> None

Register standard FastAPI / Starlette exception handlers for StreamLeaseRejected (429) and StreamLeaseUnavailable (503).

stream async

stream(
    user_id: str | int,
    stream: AsyncIterable[Any],
    media_type: str = "text/event-stream",
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    auto_renew: bool = True,
    renew_interval: float | None = None,
    close_source: bool = True,
    acquire_timeout: float | None = None,
    include_lease_headers: bool | None = None,
    **kwargs: Any,
) -> Any

Acquire a lease and return a protected Starlette/FastAPI StreamingResponse.

Provides 1-line streaming integration

@app.get("/stream") async def stream_view(user_id: str = Depends(auth)): return await manager.stream(user_id, token_generator())

If lease acquisition fails (e.g. 429 Too Many Requests or 503 Unavailable), an exception is raised immediately. If stream setup fails before returning, the acquired lease is safely released to prevent lingering ghost leases.

verify_cluster_config async

verify_cluster_config(
    strict: bool = False,
    retry_attempts: int = 3,
    retry_delay: float = 0.1,
) -> bool

Verify that this worker's configuration matches cluster configuration in Redis.

If no configuration is registered yet, this worker's fingerprint is recorded atomically. If a mismatch is detected: - If strict=True: raises ConfigurationMismatchError. - If strict=False: logs a warning and returns False.

Transient network errors during Sentinel election or network blips are retried up to retry_attempts times (spaced by retry_delay). If the coordination backend remains unavailable past all retry attempts: - If strict=True: raises StreamLeaseUnavailable (fail-fast on k8s CrashLoopBackOff). - If strict=False: logs a warning, triggers on_backend_error, and returns False.

Returns:

Name	Type	Description
bool	bool	True if configuration matches or was registered; False otherwise.

Lease

StreamLease dataclass

StreamLease(
    lease_id: str,
    user_id: str | int,
    user_key: str,
    global_key: str,
    manager: StreamLeaseManager,
    created_at: float = time.time(),
    created_monotonic: float = time.monotonic(),
)

Represents an active, acquired stream lease.

cancel_renewal

cancel_renewal() -> None

Immediately cancel all active auto-renewal background tasks for this lease.

renew async

renew() -> bool

Manually renew this lease, extending its TTL in Redis.

Returns:

Name	Type	Description
bool	bool	True if successfully extended, False if the lease expired or was evicted.

Raises: StreamLeaseUnavailable: If the Redis backend is unreachable.

release async

release(reason: str = 'manual') -> None

Explicitly release this lease from Redis.

wrap async

wrap(
    stream: AsyncIterable[T],
    auto_renew: bool = True,
    renew_interval: float | None = None,
    close_source: bool = True,
) -> AsyncIterator[T]

Wrap an async stream (e.g. SSE event generator or LLM token stream).

Renew during long pauses and release when the iterator closes. If renewal fails, interrupt the stream with StreamLeaseLost. When close_source=True, deterministically invokes aclose() (or close()) on the underlying stream upon completion, early termination, cancellation, or error.

as_streaming_response

as_streaming_response(
    stream: AsyncIterable[Any],
    media_type: str = "text/event-stream",
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    auto_renew: bool = True,
    renew_interval: float | None = None,
    close_source: bool = True,
    include_lease_headers: bool | None = None,
    **kwargs: Any,
) -> Any

Wrap an async stream and return a Starlette/FastAPI StreamingResponse.

Automatically manages lease renewal during iteration and releases Redis resources when the stream closes or client disconnects.

Configuration

LeaseConfig dataclass

LeaseConfig(
    lease_seconds: float = 30.0,
    max_per_user: int = 3,
    max_global: int = 500,
    key_prefix: str = "stream_lease",
    fail_open: bool = False,
    retry_after_seconds: int = 5,
    acquire_timeout: float = 0.0,
    on_acquired: Any = None,
    on_rejected: Any = None,
    on_lost: Any = None,
    on_released: Any = None,
    on_backend_error: Any = None,
    hook_queue_size: int = 1024,
    failure_policy: BackendFailurePolicy | None = None,
    upstream_cleanup_timeout: float = 2.0,
    telemetry_scope: str | None = None,
    include_lease_headers: bool = False,
)

Configuration for stream lease management.

lease_seconds class-attribute instance-attribute

lease_seconds: float = 30.0

Duration (in seconds) of each lease before expiring in Redis if not renewed.

max_per_user class-attribute instance-attribute

max_per_user: int = 3

Maximum concurrent streams allowed for a single user/principal key.

max_global class-attribute instance-attribute

max_global: int = 500

Maximum concurrent streams allowed across the entire cluster.

key_prefix class-attribute instance-attribute

key_prefix: str = 'stream_lease'

Prefix for Redis keys (e.g. stream_lease:user:{id}, stream_lease:global).

fail_open class-attribute instance-attribute

fail_open: bool = False

If True, allows streams to proceed unthrottled with an emergency stub if Redis is down.

retry_after_seconds class-attribute instance-attribute

retry_after_seconds: int = 5

Default Retry-After header value (in seconds) for HTTP 503 responses.

acquire_timeout class-attribute instance-attribute

acquire_timeout: float = 0.0

Duration (in seconds) to wait for an available lease slot before rejecting acquisition.

on_acquired class-attribute instance-attribute

on_acquired: Any = None

Optional callback hook: on_acquired(lease: StreamLease) -> None | Awaitable[None]

on_rejected class-attribute instance-attribute

on_rejected: Any = None

Optional callback hook: on_rejected(user_id, reason) -> None | Awaitable[None]

on_lost class-attribute instance-attribute

on_lost: Any = None

Optional callback hook: on_lost(lease: StreamLease, reason: str) -> None | Awaitable[None]

on_released class-attribute instance-attribute

on_released: Any = None

Optional callback hook: on_released(lease, reason) -> None | Awaitable[None]

on_backend_error class-attribute instance-attribute

on_backend_error: Any = None

Optional callback hook: on_backend_error(exc: Exception) -> None | Awaitable[None]

hook_queue_size class-attribute instance-attribute

hook_queue_size: int = 1024

Maximum capacity of the background hook queue before dropping telemetry events.

failure_policy class-attribute instance-attribute

failure_policy: BackendFailurePolicy | None = None

Encapsulated failure degradation policy and circuit breaker configuration.

upstream_cleanup_timeout class-attribute instance-attribute

upstream_cleanup_timeout: float = 2.0

Maximum duration (in seconds) to wait for upstream stream aclose() or close() cleanup before releasing the Redis lease. For synchronous close methods offloaded to worker threads, this bounds how long teardown waits, but cannot forcibly kill threads already executing in the worker pool.

telemetry_scope class-attribute instance-attribute

telemetry_scope: str | None = None

Optional static, low-cardinality telemetry scope identifier for isolating metrics across multiple managers sharing a telemetry registry (e.g. 'llm_heavy', 'worker_alpha'). Must not contain high-cardinality dynamic data (e.g. user IDs). Defaults to key_prefix.

include_lease_headers class-attribute instance-attribute

include_lease_headers: bool = False

Whether to automatically inject X-Stream-Lease-Id, X-Stream-Lease-User-Limit, and X-Stream-Lease-Global-Limit headers into responses.

redis_ttl property

redis_ttl: int

TTL set on Redis keys to ensure dead keys self-clean (twice lease duration).

config_key property

config_key: str

Redis key for storing and verifying cluster configuration fingerprint.

effective_failure_policy property

effective_failure_policy: BackendFailurePolicy

Effective failure policy. Returns explicit failure_policy if set, or synthesizes a policy matching fail_open if failure_policy is None.

fingerprint_dict

fingerprint_dict() -> dict[str, Any]

Return critical cluster configuration parameters for consistency verification.

Circuit breaker

CircuitBreakerConfig dataclass

CircuitBreakerConfig(
    failure_threshold: int = 5,
    recovery_timeout: float = 10.0,
    jitter: float = 1.0,
    half_open_max_probes: int = 1,
)

Configuration tuning for the worker-local circuit breaker.

failure_threshold class-attribute instance-attribute

failure_threshold: int = 5

Number of consecutive transient failures before opening the circuit.

recovery_timeout class-attribute instance-attribute

recovery_timeout: float = 10.0

Seconds to wait before allowing probe requests in HALF_OPEN state.

jitter class-attribute instance-attribute

jitter: float = 1.0

Maximum random jitter (in seconds) added to recovery_timeout to prevent herd probes.

half_open_max_probes class-attribute instance-attribute

half_open_max_probes: int = 1

Maximum concurrent probe requests permitted during HALF_OPEN testing.

BackendFailurePolicy dataclass

BackendFailurePolicy(
    circuit_breaker: CircuitBreakerConfig | None = None,
    fallback_mode: FallbackMode = FallbackMode.FAIL_CLOSED,
)

Encapsulated failure degradation policy and circuit breaker configuration.

circuit_breaker class-attribute instance-attribute

circuit_breaker: CircuitBreakerConfig | None = None

Optional circuit breaker settings. If None, circuit breaker is disabled.

fallback_mode class-attribute instance-attribute

fallback_mode: FallbackMode = FallbackMode.FAIL_CLOSED

Degradation mode when Redis fails or circuit breaker is OPEN.

FallbackMode

Bases: str, Enum

Degradation policies when the coordination backend or circuit breaker is unavailable.

CircuitState

Bases: str, Enum

Operational states of the worker-local circuit breaker.

Exceptions

StreamLeaseError

Bases: Exception

Base exception for all stream lease errors.

as_response

as_response() -> Any

Convert into a Starlette/FastAPI JSONResponse if supported by the exception subclass.

as_http_exception

as_http_exception() -> Any

Convert into a Starlette/FastAPI HTTPException if supported by the exception subclass.

StreamLeaseRejected

StreamLeaseRejected(
    reason: Literal["user_limit", "global_limit"],
    retry_after: int = 5,
    detail: str | None = None,
)

Bases: StreamLeaseError

Raised when a stream lease request is rejected due to concurrency limits.

as_response

as_response() -> Any

Convert this rejection into a JSONResponse (HTTP 429).

as_http_exception

as_http_exception() -> Any

Convert this rejection into a FastAPI/Starlette HTTPException (HTTP 429) to be raised.

StreamLeaseUnavailable

StreamLeaseUnavailable(
    retry_after: int = 5, detail: str | None = None
)

Bases: StreamLeaseError

Raised when the Redis backend is unavailable to coordinate stream leases.

as_response

as_response() -> Any

Convert into a JSONResponse (HTTP 503).

as_http_exception

as_http_exception() -> Any

Convert into a FastAPI/Starlette HTTPException (HTTP 503).

StreamLeaseLost

StreamLeaseLost(lease_id: str)

Bases: StreamLeaseError

Raised when an active stream can no longer hold its concurrency slot.

ConfigurationMismatchError

ConfigurationMismatchError(
    message: str,
    existing_config: dict[str, Any],
    current_config: dict[str, Any],
)

Bases: StreamLeaseError

Raised when worker configuration conflicts with existing cluster configuration.

Helpers

register_exception_handlers

register_exception_handlers(app: Any) -> None

Register standard FastAPI / Starlette exception handlers for StreamLeaseRejected (429) and StreamLeaseUnavailable (503).

ProtectedStreamingResponse

Bases: StreamingResponse

FastAPI/Starlette StreamingResponse subclass ensuring deterministic body_iterator aclose() cleanup on ASGI client disconnect, cancellation, or error.