Metadata-Version: 2.5
Name: fastapi-stream-lease
Version: 0.1.4
Summary: Distributed stream and SSE concurrency lease manager for FastAPI and Starlette backed by atomic Redis Lua scripts.
Project-URL: Homepage, https://github.com/agustin18/fastapi-stream-lease
Project-URL: Repository, https://github.com/agustin18/fastapi-stream-lease
Project-URL: Issues, https://github.com/agustin18/fastapi-stream-lease/issues
Author-email: Agustin Saiz <agustinsaiz02@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: concurrency,fastapi,llm-streaming,rate-limiting,redis,server-sent-events,sse,starlette,streaming
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: redis>=5.0.0
Provides-Extra: dev
Requires-Dist: fakeredis[lua]>=2.20.0; extra == 'dev'
Requires-Dist: fastapi>=0.100.0; extra == 'dev'
Requires-Dist: httpx>=0.25.0; extra == 'dev'
Requires-Dist: mypy>=1.10.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest-cov>=5.0.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.4.0; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100.0; extra == 'fastapi'
Description-Content-Type: text/markdown

# fastapi-stream-lease

[![CI](https://github.com/agustin18/fastapi-stream-lease/actions/workflows/ci.yml/badge.svg)](https://github.com/agustin18/fastapi-stream-lease/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/fastapi-stream-lease.svg)](https://pypi.org/project/fastapi-stream-lease/)
[![Python](https://img.shields.io/pypi/pyversions/fastapi-stream-lease.svg)](https://pypi.org/project/fastapi-stream-lease/)

Limit **simultaneously active** SSE, LLM token streams, and WebSocket sessions across FastAPI/Starlette workers with Redis. Set a per-user limit, a global limit, or both. An extra connection receives HTTP 429 before streaming begins.

Request rate limiters answer “how many requests arrived this minute?” This package answers “how many streams are open right now?” It is useful when a connection can remain open for seconds or minutes. For a general-purpose distributed semaphore, consider [py-redis-limiters](https://pypi.org/project/redis-limiters/) or [py-redis-semaphore](https://pypi.org/project/py-redis-semaphore/).

## Install

Requires Python 3.10+ and Redis 5.0+.

With Redis 5 and redis-py 8+, construct your client with `redis.from_url(url, protocol=2)`: redis-py 8 defaults to RESP3, which Redis 5 does not support. See the [redis-py protocol documentation](https://github.com/redis/redis-py#resp3-support).

```bash
pip install 'fastapi-stream-lease[fastapi]'
```

The `fastapi` extra supplies FastAPI for the HTTP helpers; the core package only requires `redis`.

## Try it locally

The [runnable SSE example](examples/sse_demo.py) accepts one API key from `STREAM_DEMO_TOKEN` and uses it as a demo identity. Start Redis, clone this repository, then run:

```bash
git clone https://github.com/agustin18/fastapi-stream-lease.git
cd fastapi-stream-lease
pip install -e '.[fastapi]' uvicorn
export STREAM_DEMO_TOKEN=local-secret
export REDIS_URL=redis://localhost:6379/0
uvicorn examples.sse_demo:app
```

In another terminal, open two streams, then try a third with the same key:

```bash
curl -N -H 'X-API-Key: local-secret' http://localhost:8000/stream
```

The first two requests stream events; the third receives `429` with a `Retry-After` header until a slot is freed. Replace the demo API key with your application's authenticated user or account ID before production use. Never use an unverified request parameter as the user ID.

## FastAPI integration

```python
from fastapi import Depends, FastAPI, Request
from fastapi.responses import StreamingResponse
import redis.asyncio as redis

from fastapi_stream_lease import (
    LeaseConfig,
    StreamLeaseManager,
    StreamLeaseRejected,
    StreamLeaseUnavailable,
)

app = FastAPI()
# Set explicit timeouts so slow Redis calls do not block worker threads
redis_client = redis.from_url(
    "redis://localhost:6379",
    socket_timeout=1.0,
    socket_connect_timeout=1.0,
)
manager = StreamLeaseManager(
    redis_client,
    LeaseConfig(max_per_user=2, max_global=500, lease_seconds=30),
)


async def authenticated_user_id() -> str:
    # Illustrative placeholder: replace with your auth dependency (e.g. API key or JWT sub).
    # For fully runnable code, see examples/sse_demo.py and examples/websocket_demo.py.
    return "user-123"


@app.exception_handler(StreamLeaseRejected)
async def rejected(request: Request, exc: StreamLeaseRejected):
    return exc.as_response()


@app.exception_handler(StreamLeaseUnavailable)
async def unavailable(request: Request, exc: StreamLeaseUnavailable):
    return exc.as_response()


@app.get("/stream")
async def stream(user_id: str = Depends(authenticated_user_id)):
    lease = await manager.acquire(user_id)

    async def events():
        yield "data: first event\n\n"
        # Yield tokens or events here...

    return StreamingResponse(lease.wrap(events()), media_type="text/event-stream")
```

Close the Redis client in your application's lifespan shutdown handler. Reuse the same `StreamLeaseManager` and `key_prefix` across workers that share limits. `max_per_user=0` or `max_global=0` disables that limit; the global count is unavailable when global tracking is disabled.

For WebSockets, keep the context open for the whole session:

```python
async with manager.lease(user_id) as lease:
    while True:
        message = await websocket.receive_text()
        await websocket.send_text(message)
```

The manager context and `async with lease` both renew while open. Handle normal WebSocket disconnects in your route as usual. If renewal fails or the lease expires, `StreamLeaseLost` interrupts the stream or context. Catch it at the application boundary if you want to record a metric or send an application-specific WebSocket close code. `wrap(auto_renew=False)` disables automatic renewal; use it only if you renew the lease yourself.

## Behavior and limits

- Acquisition, renewal, expiration cleanup, and release use atomic Redis Lua scripts. The keys share a Redis Cluster hash tag, so a user and global limit can be checked in one script.
- Every lease expires after `lease_seconds` without a successful renewal. `wrap()` and `manager.lease()` renew every half interval by default. Transient Redis connection errors trigger fast retries across the remaining lease TTL (Adaptive Grace Period), preventing temporary hiccups from dropping active streams.
- If Redis is unavailable on initial acquisition, `StreamLeaseUnavailable` (HTTP 503) is raised by default (`fail_open=False`). Set `fail_open=True` in `LeaseConfig` if your application prefers allowing streams during Redis outages (graceful degradation).
- Normal completion or cancellation attempts immediate release. If Redis is unavailable during release, the lease is removed after expiration; cleanup of the key itself uses a longer TTL. An async iterator abandoned without being closed may also hold its slot until expiration. Use `contextlib.aclosing()` if your own consumer stops iteration early.
- `get_active_count(user_id)` counts active leases for one identity; `get_active_count()` counts globally when `max_global` is enabled. Neither is a historical usage metric.
- All workers sharing limits must use the same key prefix and compatible limit settings. Lease expiration is measured by Redis, avoiding clock differences among application workers.

## Production and Operational Guide

- **Redis Client Timeouts:** Always configure explicit timeouts on your Redis client (e.g. `socket_timeout=1.0, socket_connect_timeout=1.0`). Without timeouts, an unreachable Redis instance can block asyncio event loop execution indefinitely.
- **Fail-Open vs. Fail-Closed Strategy:**
  - `fail_open=False` (Default): Raises `StreamLeaseUnavailable` (HTTP 503) when Redis is unreachable. Enforces limits during transient network partitions at the cost of rejecting requests when the backend is down. (Note: asynchronous Redis replication or master failover can still lose recently acknowledged writes if a master fails before syncing to its replica).
  - `fail_open=True`: Automatically grants in-memory fallback leases when Redis encounters network or timeout errors. Keeps streaming endpoints open during outages, with the operational trade-off that limits are not coordinated across workers until Redis recovers. Authentication, authorization, and script syntax errors never fail open.
- **Definitive Revocation vs. Network Errors:** If Redis explicitly reports that a lease is missing or expired (`renew()` returning 0) or encounters an unhandled execution error, `wrap()` and `lease()` cancel the stream immediately to prevent exceeding limits. Transient network disconnects trigger rapid retries until the monotonic lease deadline is reached.
- **Observability and Lifecycle Hooks:**
  `LeaseConfig` provides zero-dependency callback hooks (supporting both sync and async callables) to plug directly into Prometheus, Datadog, StatsD, or Sentry:
  ```python
  config = LeaseConfig(
      on_acquired=lambda lease: PROMETHEUS_ACQUIRED.inc(),
      on_rejected=lambda uid, reason: PROMETHEUS_REJECTED.labels(reason=reason).inc(),
      on_lost=lambda lease, reason: PROMETHEUS_LOST.labels(reason=reason).inc(),
      on_backend_error=lambda exc: PROMETHEUS_BACKEND_ERRORS.inc(),
  )
  ```
- **Redis Cluster:** All keys use Redis hash tags (`{prefix}:user:...` and `{prefix}:global`), guaranteeing user and global sorted sets reside on the same hash slot for multi-key atomic Lua operations. As with any multi-key Lua coordination, evaluate slot contention and failover behavior under your specific topology.

## Contributing and security

See [CONTRIBUTING.md](CONTRIBUTING.md) for the local workflow and [SECURITY.md](SECURITY.md) for private vulnerability reports. Changes are proposed through pull requests and merged by the maintainer after CI passes. The package is licensed under [MIT](LICENSE).

CI checks formatting, lint, types, Redis 5 and 7 behavior, package build, and a minimum of 95% combined line and branch coverage. This is a small beta project; reports from real deployments are especially helpful for documenting operational limits.
