scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-06 07:48:06 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 112

refactor(redis): plain-method transport, store layer and response polling store

Browse source 
Rework RedisTransport into plain get, mget, set, delete and eval methods,
split the breaker step (run_under_breaker over the existing
_run_under_circuit_breaker) from the events step (RedisServiceEvents.observe,
which skips the stack walk and event when no service_callback is set), and
add eval with EVALSHA plus EVAL fallback. Fold in the store layer and the
response polling store, and replace the redis-access skill references with
the migration plan and PR sketches.
This commit is contained in:
Yujong Lee 2026-10-29 23:27:00 -07:00
parent d04bf18333
commit a6202af364
19 changed files with 1385 additions and 489 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

160
.agents/skills/redis-access/SKILL.md
View file

@ -1,17 +1,163 @@
---
name: redis-access
description: Add Redis operations or move raw redis-py clients and scripts behind RedisTransport in litellm/redis, without porting them to Rust or expanding migration scope
description: Add Redis operations, move raw redis-py access or a keyspace behind RedisTransport and a store in litellm/redis, or explicitly port a store to Rust, without mixing the Python and Rust Redis stacks or expanding migration scope
metadata:
internal: true
---
# Redis access
Python reaches Redis through `RedisTransport` (`litellm/redis/transport/contract.py`). Read `references/transport.md` for adding a command, the failure kinds, the namespace and TTL rules, and moving raw `init_async_client` / `async_register_script` / `.redis_client` access onto the transport
## Two stacks, never mixed
Rules that always apply
- Python keyspace code reaches Redis one way only, `Python store -> RedisPyTransport -> redis-py`
- Rust keyspace code reaches Redis one way only, `Python caller -> pyo3 -> crates/redis-store -> crates/redis-client`, and the Rust gateway enters at `crates/redis-store`
- A Python store never runs on the Rust client, and a Rust store never goes back through the Python transport. There is no Rust-backed `RedisTransport`, since that would be a second redis client binding over pyo3 that has to grow toward redis-py's whole command set
- A keyspace is ported as a unit, keys plus codec plus Lua plus client, or not at all. Which stack a keyspace is on is decided per keyspace, never per call
- During the transition a pod holds one redis-py pool and one Rust client to the same Redis, with one breaker each. That is correct, they guard different sockets. redis-py goes away only when the last keyspace, pub/sub and admin have a Rust home
- New code never calls `init_async_client`, `init_pubsub_client`, `async_register_script` or `.redis_client`, and never imports redis-py outside `litellm/redis/transport/redis_py.py`
- Keys and values in Redis are a wire format shared with older pods and Rust. A move keeps them byte for byte
- A `transport` failure on a write is an unknown outcome, never "not applied"
- Do only what was asked. Confirm before any Rust work
## Adding new Redis access
- New coordination state gets a store in `litellm/redis/stores/<keyspace>.py`, not a new `RedisCache` method
- A Python store is a finished state. Adding one does not require a Rust port
- Callers get typed domain results from the store and keep policy, which TTL applies, fail open or closed, retries, and the `with_service_target` that labels the calls
- New code never calls `init_async_client`, `init_pubsub_client`, `async_register_script` or `.redis_client`, and never imports redis-py outside `litellm/redis/transport/redis_py.py`. `litellm/caching/redis_cache.py` and `redis_cluster_cache.py` stay allowlisted while they own the breaker's exception types and the generic verbs
## The transport
`RedisTransport` in `litellm/redis/transport/contract.py` is how Python keyspace code reaches redis-py. It is a Protocol of plain methods, one per Redis command the Python code uses (`get`, `mget`, `set`, `delete`, `eval`), command-shaped and never redis-py shaped
- Every method returns its value or a `RedisFailure`. Nothing raises for a Redis problem
- `RedisFailure.kind` is `transport` (timeout or lost connection, the command may still have applied, so a failed write is an unknown outcome, never "not applied"), `server` (Redis answered with an error, the connection is fine and the breaker does not count it), `decode` (the reply was not the shape expected) or `breaker_open` (refused without touching the socket)
- The namespace is applied once, inside the transport, to every key including Lua `KEYS`. A key that already starts with `{namespace}:` is left alone. Stores and callers pass logical keys
- Every write takes its TTL from the caller as a `timedelta`. `None` means no expiry, there is no default
- Replies are undecoded bytes. `eval` returns `int | bytes | None | tuple[...]`. Decoding belongs to the store that owns the keyspace
- `eval` takes script text every call and does `EVALSHA` with `EVAL` fallback, so nothing registers scripts
- These rules are shared with `crates/redis-client`, whose `AGENTS.md` has the Rust side. Change both together, since Python and Rust pods share keys in one Redis
- `RedisCache.transport()` builds the transport over that cache's own client, namespace, circuit breaker, `_admit_async_call` and `ServiceLogging`. No new connection, no second breaker, and `RedisClusterCache`'s connection gate still applies
- Every method runs its command through two steps, in this order. `run_under_breaker` wraps the existing `_run_under_circuit_breaker` from `litellm/caching/redis_cache.py`, so admission, the generation check, the health failure classification and the timeout flag are the ones legacy traffic uses, and maps the outcome to a `RedisFailure` (`breaker_open`, `transport`, `server`) instead of raising. `RedisServiceEvents.observe` times the call, logs a failure and emits the service event
```python
@dataclass(frozen=True, slots=True)
class RedisPyTransport:
client: Callable[[], AsyncRedisCommands]
namespace: str | None
breaker: RedisCircuitBreaker
admit: Admit
events: RedisServiceEvents
async def get(self, key: str) -> bytes | None | RedisFailure:
return _bytes_reply(await self._run("get", lambda: self.client().get(self._key(key))))
async def _run(self, name: str, call: Callable[[], Awaitable[_T]]) -> _T | RedisFailure:
return await self.events.observe(name, lambda: run_under_breaker(self.breaker, name, self.admit, call))
```
- Adding a command means a method on the Protocol, an implementation in `redis_py.py`, and a fakeredis test that the namespace is applied once, the TTL lands, a rejected command is `server` and an outage is `transport`. Before adding one, check whether the operation is really one `eval`. A counter with a TTL on first write or an owner-checked release is one script, not three round trips
- Multi-key commands and scripts keep their keys in one slot, or they break on cluster depending on slot ownership
## Stores
A store owns one keyspace. Python and Rust pods share one Redis, so each keyspace needs one definition of its keys and values, and the store is that definition
- One module per keyspace. It defines a `Protocol` of domain operations named for what the caller wants (`acquire`, `release`, `increment`, `get`), never for Redis commands, and a frozen dataclass implementation that takes a `RedisTransport`
- The store owns, and nothing outside it knows, the key builders (logical keys, no namespace), the value codec, expiry taken from the caller, the Lua, and reply decoding into typed results
- Every operation returns its result or a `RedisFailure`. A value that does not decode is a `decode` failure, never a miss or a default
- No logging, no retries, no fallback in a store. Those are policy and stay in the caller
- A caller never builds a key, touches a client or runs a command for a keyspace that has a store. Exposing a key builder as a delegate for an existing public helper is fine
- Keys and values already in Redis are a wire format. Old and new pods read each other's writes during a rolling deploy, so a move keeps them byte for byte, including `json.dumps` spacing and ASCII escaping, float formatting, and seconds versus milliseconds. Changing a key, codec or TTL unit is never part of a move. It is its own change that writes a new versioned key and reads both for one release
- Response cache identity (normalization, hashing into a logical key) is backend independent and is not store logic. Reuse or move the Rust `cache_response::cache_key`, never write a third key builder
## Asked to move existing access behind a store
- Move the key builders, codec, TTLs, Lua and decoding verbatim. Keep policy in the caller
- Keep the caller's outcome per failure. If it failed closed on any exception, it fails closed on any `RedisFailure`
- Delete the caller's own timing, service hooks and Redis health logging. The transport emits for it now
- Move one keyspace per change. An extraction is not a Rust port
- `RedisCache` is a connection object wearing a cache name. Its own cache verbs move onto the transport one method at a time, each conversion deleting that method's inline logging, until what is left is the connection, the breaker and a thin facade
- Not on the transport yet, so leave them on `RedisCache` until the transport grows them: pub/sub, admin commands, pipelines and `RedisBatch`. Vector and semantic search keep their own clients
## Observability
`ServiceLogging` (`litellm/_service_logger.py`) is a shared infra-health channel with many producers (Redis, Postgres, the spend queues, auth, router) and three sinks (`prometheus_system`, `datadog`, OTel). Redis work only decides where the Redis producer lives, never touches the sink side
- `RedisServiceEvents` (`litellm/redis/transport/events.py`, a frozen dataclass holding the `ServiceLogging`) is the only place REDIS service events for Python traffic are emitted. Its one method, `observe(call_type, call)`, times the call, logs a failure with the caller's service target, emits one success or failure event in the caller's task, and returns the result unchanged. The transport calls it with the method name as `call_type`. Stores and callers emit nothing
- When no `service_callback` is configured, which is the default, `observe` runs the call and logs the failure without walking the stack or building an event
- Rust never calls back into Python per Redis operation. A per-op host call costs a tokio task, a blocking thread, the GIL and an event loop per op, and buffering plus a flusher loses the OTel parent and needs a background task the SDK does not have
- A native store holds the same `RedisServiceEvents` and runs each operation through `observe`, so native traffic reports from Python exactly like transport traffic, with the store operation name as `call_type`
- The breaker for the native client is `RedisConfig::breaker` in Rust. A native store never holds a Python `RedisCircuitBreaker`, one breaker per socket. Breaker transitions are rare and may reach Python through one host call that folds the Rust breaker's state into `redis_breaker_state_counts()`
- Keep metric names and label names as they are. Dashboards alert on `litellm_redis_*`. The one accepted label value change is `call_type`, which becomes the transport method name when a keyspace moves, and `prometheus_services.py` writes it into `function_name`
## Asked to port a store to Rust
Confirm with the user before the first port, since the bridge for it does not exist yet
- The keyspace's module in `litellm-rust/crates/redis-store` implements the same operations as async fns on `crates/redis-client`. Its `AGENTS.md` has the crate rules. Rust callers use the same module, so there is one implementation per operation
- One `#[pyclass]` per store in `crates/python-bridge/src/redis/`, built on one shared native client handle, so Python opens no extra connection per store
- A `Native<Keyspace>Store` in the Python module implements the same `Protocol` over the pyclass, each operation through `RedisServiceEvents.observe`, so callers do not change. The implementation is picked at construction, and the transport-backed one stays for config the native client does not support yet (Sentinel, cloud auth, TLS cert options)
- Results come back as frozen Pydantic models generated from the Rust types, so callers keep matching on the same classes. A decode error maps to `RedisFailure(kind="decode")`
- The codec writes the exact bytes Python writes. `json.dumps` escapes non-ASCII and `serde_json` does not, so a Rust pod could not release a lock a Python pod took
- Each port ships cross-implementation tests against one Redis, Python writes and Rust reads and the reverse, covering keys, values, expiry and atomicity through a rolling deploy
## Tests
- Test a store through its caller against `fake_redis_cache(server)` from `tests/_support/fake_redis.py`, a real `RedisCache` on a `fakeredis.FakeServer`, so namespace, breaker and failure mapping are the production ones. No mocks of our own code
- Cover each result the caller acts on, that the TTL lands (`pttl` on a raw `FakeAsyncRedis`), and an outage (`server.connected = False`)
- When a keyspace moves, seed with the old writer and read with the new reader, and the reverse, so old and new pods interoperate
- Assert returned values and Redis state, never the commands sent
- Rust store tests run against real Redis from `litellm-redis-testing` on both topologies
## Decisions from the first slices
The first transport and store slices settled most of the shape. Keep what is listed under "Keep" as is, and rework anything that still has a "Change" item before extending it
Keep
- `RedisCache.transport()` builds the transport over the cache's own client, namespace and breaker, no second connection, no second breaker
- `RedisFailure` as a value with kinds `transport`, `server`, `decode`, `breaker_open` and a `timeout` flag, and a failed write as an unknown outcome
- Namespace applied once inside the transport, TTL from the caller as a `timedelta` sent in milliseconds, replies as undecoded bytes
- The two steps around every redis-py command, the breaker and the events, one REDIS service event per command, none for `breaker_open`, one WARNING per failure through the timeout throttle with the caller's `with_service_target` in the line
- `call_type` becoming the transport method name (`get`, `set`, `delete`) when a caller moves off `async_*_cache`
- The store shape, a `Protocol` of domain operations plus a frozen dataclass over the transport that owns keys, codec, expiry, Lua and decoding, with policy in the caller
- Byte for byte wire compatibility on a move, including the response polling double `json.dumps`, proven by seeding with the old writer and reading with the old reader
- Tests through the caller on `fake_redis_cache(server)`, asserting values and Redis state, never commands sent
- The Rust store crate shape, async fns on `crates/redis-client` returning domain types, one pyclass per store, a `Native<Keyspace>Store` behind the same `Protocol` picked at construction, frozen Pydantic results, cross-implementation tests on one Redis
Change
- Command dataclasses (`Get`, `Set`, `Delete`) run through an overloaded `execute` with a `match` per backend. With redis-py as the only backend they only add a second place to edit per command. The transport is plain methods
```python
# before
await transport.execute(Set(key, value, ttl=ttl, nx=True))
await transport.execute(Get(key))
# after
await transport.set(key, value, ttl=ttl, nx=True)
await transport.get(key)
```
- A pyo3 `NativeRedisTransport` over the Rust client behind the same Protocol. It mixes the stacks and would have to grow toward redis-py's command set. The Rust path is store level only
- "Native stores skip the Python breaker and events, so both have to come from Rust first". Half right. The breaker comes from Rust, the events do not
```python
@dataclass(frozen=True, slots=True)
class NativeLockStore:
native: NativeLockStoreHandle # pyclass on the shared Rust client, breaker inside
events: RedisServiceEvents # no Python breaker here, one breaker per socket
async def acquire(self, key: str, owner: str, *, ttl: timedelta) -> bool | RedisFailure:
return await self.events.observe("acquire", lambda: _reply(self.native.acquire(key, owner, ttl_ms(ttl))))
```
- A Rust `RedisEvents` implementation that reports every native call to `ServiceLogging` through a host call. Delete the per-call path and keep only the breaker transition
- A transport framed as "the backend swaps to the Rust client without touching callers". The transport is the redis-py seam, nothing else goes behind it
- The events step walking the stack for `caller` and spawning a task per command whether or not a sink is configured. Check `ServiceLogging.has_sinks()` first
- The names `Guard`, `BreakerGuard` and `ReportingGuard`, inherited from the `_redis_circuit_breaker_guard` decorator. Only the breaker guards anything. The breaker step is a function, `run_under_breaker(breaker, name, admit, call)`, which wraps `_run_under_circuit_breaker` and maps its exceptions to `RedisFailure`, and the events step is `RedisServiceEvents.observe`. The transport holds the breaker, the cache's `_admit_async_call` and the events as fields and there is no generic wrapper Protocol between them. Reimplementing the breaker admission in the transport, with `allow()` and a bare `record_failure()`, drops the generation check and the timeout class and is not an option
- Lua through `async_register_script`, a per-loop executor cached in `in_memory_llm_clients_cache`. `transport.eval` takes the script text and does `EVALSHA` with `EVAL` fallback, no registration and no per-loop state
- Lists in replies. `mget` returns a tuple with one slot per key and `eval` returns `int | bytes | None | tuple[...]`, a Lua `false` as `None`
## Scope
- The milestone order for the migration, and what each PR may and may not touch, is in `references/migration-plan.md`. The first two milestones are Python only and ship first. Nothing under `litellm-rust/` changes until milestone 2 has merged
- A keyspace moves whole in one commit, every operation the caller runs on those keys. Moving the read and leaving the write on `RedisCache` is not a move
- Do only what was asked. An extraction is not a store move, a store move is not a Rust port, and a Rust port of one keyspace does not switch another
- A Rust store that already exists does not by itself justify switching its Python caller. Switching callers is its own change with live verification

76
.agents/skills/redis-access/references/migration-plan.md Normal file
View file

@ -0,0 +1,76 @@
# Redis migration plan
Two Python PRs first, then the Rust work. Milestones 1 and 2 are Python refactors that put every raw redis-py keyspace behind a store on the transport, so a keyspace can later move to Rust without touching its callers. They are not behavior free. Each lists its accepted behavior changes, and nothing outside that list may change. Milestones 3 and 4 are the Rust side and start only after milestone 2 merges, nothing in 1 or 2 depends on them or touches `litellm-rust/`
Every keyspace port after milestone 4 repeats its pattern, one keyspace at a time. `pr-sketches.md` has the interfaces each milestone brings
## 1. Transport, store layer and response polling (PR 1)
Python only
- `litellm/redis/transport/` with `RedisTransport`, `RedisPyTransport`, `RedisFailure`, `run_under_breaker` and `RedisServiceEvents`, built through `RedisCache.transport()`
- `run_under_breaker` wraps the existing `_run_under_circuit_breaker` in `litellm/caching/redis_cache.py`. Admission, the generation check, the swallowed-failure check, `_is_redis_health_failure` and the timeout flag stay exactly what `RedisCache` has today, the function only maps the outcome to a `RedisFailure`. It does not reimplement the breaker
- The transport takes the cache's `_admit_async_call`, so `RedisClusterCache`'s connection gate still applies to every command on the transport path
- Every "Change" item in `SKILL.md` applied before merge, plain methods instead of command dataclasses, `has_sinks()` checked before the events step walks the stack, `eval` taking script text with `EVALSHA` and `EVAL` fallback, tuples in replies
- `litellm/redis/stores/` with one keyspace moved, response polling, seeded by the old writer and read by the new reader and the reverse. One keyspace is enough to prove the shape, and it is the only one of the first slices that moves whole without a new Lua script
- Folds 48958, 48956, and from 49162 only `litellm/redis/stores/__init__.py` and the docs. The `litellm-rust/crates/redis-store` crate in 49162 waits for milestone 4. 49163 moved only `peek` of the MCP single-use marker, half a keyspace, so it folds into milestone 2 where the marker moves whole. 49164 (native transport) is closed, it mixes the two stacks
Accepted behavior changes
- REDIS service events carry the transport method name as `call_type` (`get`, `set`, `delete`, `eval`) instead of `async_get_cache` and friends. `litellm/integrations/prometheus_services.py` puts `call_type` into the `function_name` label, so that label's values change for moved keyspaces. Metric names and the other labels stay
- A stored value that does not decode is a `decode` failure in the store, never the `ast.literal_eval` fallback that `_get_cache_logic` has
- With no `service_callback` configured the events step neither walks the stack nor builds an event. The WARNING per failure still logs
- The REDIS service event is awaited in the caller's task instead of spawned as its own task
- The initial polling state is written through `json.dumps` like every later update, so its inner document gains the default `", "` and `": "` separators and ASCII escaping that `model_dump_json()` left out. Old and new readers both parse it, and every update already wrote that form
## 2. Every remaining raw redis-py keyspace behind a store (PR 2)
Python only. One commit per keyspace so each move reviews and reverts on its own. Split `parallel_request_limiter_v3` into its own PR if the diff gets large, it is the hot path
The list is the census of modules outside `litellm/caching/` that call `init_async_client()`, `.redis_client` or `async_register_script`, minus pub/sub and admin
- MCP single-use marker in `gateway_dcr_flow.py`, `peek` and `claim` together. `claim` becomes one `eval` doing `INCRBYFLOAT` plus `EXPIRE` when the key has no TTL, the bytes `async_increment` writes today
- Pod lock manager
- `parallel_request_limiter_v3` and its Lua. It also increments through `async_increment_cache_pipeline` on `DualCache`. Before splitting, check whether that pipeline touches the keys the scripts own. If it does, the keyspace moves whole and the transport grows what the pipeline does as one `eval`. If not, the pipeline stays on the generic verbs
- Max budget per session and max iterations limiters
- Batch enqueued tokens
- Login throttle
- Affinity cache
- MCP DCR client records and the outbound credential refresh coordinator
Accepted behavior changes
- Pod lock release is the compare-and-delete Lua only. The GET plus DEL fallback goes, since `eval` already falls back from `EVALSHA` to `EVAL`. Before merge, confirm no supported Redis deployment disables scripting. If one does, the fallback stays in `PodLockManager` as caller policy, never in the store
- Each caller's `call_type` and decode changes as in milestone 1
Done when nothing outside `litellm/redis/` and `litellm/caching/` calls `.redis_client`, `init_async_client` or `async_register_script` for these keyspaces, and a lint rule bans importing redis-py outside `litellm/redis/transport/redis_py.py`. The allowlist is `litellm/caching/redis_cache.py` and `litellm/caching/redis_cluster_cache.py`, which import `redis.exceptions` for the breaker and still own the generic verbs, plus the remaining users below. The list only shrinks
Stays on redis-py and out of this plan, per `SKILL.md`
- Pub/sub (`config_sync_pubsub`)
- Admin commands (`debug_utils`)
- Pipelines and `RedisBatch`
- Semantic and vector caches, which keep their own clients
- The generic `RedisCache.async_*` verbs used through `DualCache`, which move onto the transport one method at a time as separate work
- Spend update buffer and `batch_redis_get`. They never touch redis-py directly, they use `async_rpush_pipeline`, `async_lpop_pipeline`, `async_rpush_and_trim` and `async_batch_get_cache`, so they move with the generic verbs and pipelines, not here
## 3. Rust client parity
Rust only, nothing calls the new paths yet. This is 49189, on top of 49186 (SecretValue). It does not block or depend on 1 and 2
- Script ordering in pipelines, TTL in the same script for counter writes, breaker classification per topology matching redis-py, redis-py timeout and reconnect defaults
- TLS, Sentinel and cloud auth config, so fewer deployments need the transport-backed fallback in milestone 4
## 4. Bridge and the first native store
Starts after 2 merges, needs 3
- `litellm-rust/crates/redis-store` with the pod lock keyspace as async fns on `crates/redis-client`
- One shared native client handle in `crates/python-bridge/src/redis/` and one `#[pyclass]` for the lock store on it
- The breaker transition host call folding the Rust breaker into `redis_breaker_state_counts()`
- `NativePodLockStore` behind the existing `Protocol`, each operation through `RedisServiceEvents.observe`, picked at construction, with the transport-backed store kept for config the Rust client does not support yet
- Cross-implementation tests on real Redis from `litellm-redis-testing` (`litellm-rust/crates/redis-testing`), Python writes and Rust reads and the reverse, covering keys, values, expiry and owner-checked release
Pod lock goes first because it is low traffic and its owner-checked release only works if both stacks write the same `json.dumps` encoded owner
Switching the pod lock caller to the native store is its own change with live verification against a proxy, either a follow-up PR or a separately proven final commit

432
.agents/skills/redis-access/references/pr-sketches.md Normal file
View file

@ -0,0 +1,432 @@
# PR sketches
Interface sketches for each milestone in `migration-plan.md`. They show the shape each PR brings, not finished code. Bodies are elided with `...` where they are mechanical. Names and signatures here are what reviewers should hold a PR to, and any change to them updates this file in the same PR
## 1. Transport, store layer and response polling
Python only. Accepted behavior changes are listed in `migration-plan.md`
### `litellm/redis/transport/contract.py`
```python
Arg: TypeAlias = str | bytes | int | float
EvalReply: TypeAlias = int | bytes | None | tuple["EvalReply", ...]
class Admit(Protocol):
def __call__(self, call: Callable[[], Awaitable[_T]], /) -> Awaitable[_T]: ...
@dataclass(frozen=True, slots=True)
class RedisFailure:
kind: Literal["transport", "server", "decode", "breaker_open"]
message: str
timeout: bool = False
def ttl_ms(ttl: timedelta) -> int:
return max(1, math.ceil(ttl / timedelta(milliseconds=1)))
class RedisTransport(Protocol):
async def get(self, key: str) -> bytes | None | RedisFailure: ...
async def mget(self, keys: Sequence[str]) -> tuple[bytes | None, ...] | RedisFailure: ...
async def set(self, key: str, value: Arg, *, ttl: timedelta | None, nx: bool = False) -> bool | RedisFailure: ...
async def delete(self, *keys: str) -> int | RedisFailure: ...
async def eval(self, script: str, keys: Sequence[str], args: Sequence[Arg]) -> EvalReply | RedisFailure: ...
```
### `litellm/redis/transport/breaker.py`
```python
async def run_under_breaker(
breaker: RedisCircuitBreaker, name: str, admit: Admit, call: Callable[[], Awaitable[_T]]
) -> _T | RedisFailure:
try:
return await _run_under_circuit_breaker(breaker, name, lambda: admit(call))
except RedisCircuitBreakerOpenError as e:
return RedisFailure(kind="breaker_open", message=str(e))
except Exception as e:
failure: Final = _redis_failure(e)
if failure is None:
raise
return failure
def _redis_failure(exc: BaseException) -> RedisFailure | None:
if _is_redis_health_failure(exc):
return RedisFailure(kind="transport", message=str(exc), timeout=is_redis_timeout_failure(exc))
if isinstance(exc, RedisError):
return RedisFailure(kind="server", message=str(exc))
return None
```
`_run_under_circuit_breaker`, `_is_redis_health_failure`, `is_redis_timeout_failure` and `RedisCircuitBreakerOpenError` are the ones in `litellm/caching/redis_cache.py` today, imported, not copied. That keeps the admission record, the generation check, the swallowed-failure check and the timeout class on the breaker identical for transport and legacy traffic. `admit` is the cache's `_admit_async_call`, which on `RedisClusterCache` runs the call through the cluster connection gate. Anything that is not a redis-py error propagates, since it is a bug and not a Redis outcome
### `litellm/redis/transport/events.py`
```python
@dataclass(frozen=True, slots=True)
class RedisServiceEvents:
service_logging: ServiceLogging
async def observe(self, call_type: str, call: Callable[[], Awaitable[_T | RedisFailure]]) -> _T | RedisFailure:
if not self.service_logging.has_sinks():
result: Final = await call()
_log_failure(call_type, result)
return result
...
```
`has_sinks()` is new on `ServiceLogging` and is `bool(litellm.service_callback)`. The hooks are awaited through `ServiceLogging.emit_guarded`, so a callback that raises never fails the Redis call. `...` times the call, logs a failure through the timeout throttle with the caller's `with_service_target`, and emits one success or failure REDIS event in the caller's task. No event for `breaker_open`
### `litellm/redis/transport/redis_py.py`
The only module outside `litellm/caching/` that imports redis-py
```python
@dataclass(frozen=True, slots=True)
class RedisPyTransport:
client: Callable[[], AsyncRedisCommands]
namespace: str | None
breaker: RedisCircuitBreaker
admit: Admit
events: RedisServiceEvents
async def get(self, key: str) -> bytes | None | RedisFailure:
return _bytes_reply(await self._run("get", lambda: self.client().get(self._key(key))))
async def set(self, key: str, value: Arg, *, ttl: timedelta | None, nx: bool = False) -> bool | RedisFailure:
px: Final = None if ttl is None else ttl_ms(ttl)
return _bool_reply(await self._run("set", lambda: self.client().set(self._key(key), value, px=px, nx=nx)))
async def eval(self, script: str, keys: Sequence[str], args: Sequence[Arg]) -> EvalReply | RedisFailure:
sha: Final = _sha1(script)
namespaced: Final = tuple(self._key(k) for k in keys)
return _eval_reply(await self._run("eval", lambda: _evalsha_or_eval(self.client(), script, sha, namespaced, args)))
def _key(self, key: str) -> str:
if self.namespace is None or key.startswith(f"{self.namespace}:"):
return key
return f"{self.namespace}:{key}"
async def _run(self, name: str, call: Callable[[], Awaitable[_T]]) -> _T | RedisFailure:
return await self.events.observe(name, lambda: run_under_breaker(self.breaker, name, self.admit, call))
```
`_evalsha_or_eval` sends `EVALSHA` and retries with `EVAL` on `NoScriptError`, so there is no registration and no per-loop state. `_key` is `check_and_fix_namespace` moved, same rule
### `litellm/caching/redis_cache.py`
```python
class RedisCache(BaseCache):
def transport(self) -> RedisTransport:
return RedisPyTransport(
client=self.init_async_client,
namespace=self.namespace,
breaker=self._circuit_breaker,
admit=self._admit_async_call,
events=RedisServiceEvents(self.service_logger_obj),
)
```
`RedisClusterCache` inherits it and its `_admit_async_call` override carries the connection gate onto the transport
### `litellm/redis/stores/response_polling.py`
```python
KEY_PREFIX: Final = "litellm:polling:response:"
class ResponsePollingStore(Protocol):
async def put(self, polling_id: str, state: Mapping[str, object], *, ttl: timedelta) -> None | RedisFailure: ...
async def get(self, polling_id: str) -> Mapping[str, JsonValue] | None | RedisFailure: ...
async def delete(self, polling_id: str) -> None | RedisFailure: ...
@dataclass(frozen=True, slots=True)
class TransportResponsePollingStore:
transport: RedisTransport
async def put(self, polling_id: str, state: Mapping[str, object], *, ttl: timedelta) -> None | RedisFailure:
written: Final = await self.transport.set(_key(polling_id), json.dumps(json.dumps(state)), ttl=ttl)
return written if isinstance(written, RedisFailure) else None
...
```
The double `json.dumps` is the existing wire format, kept byte for byte, and `get` undoes both. A reply that is not valid JSON twice over is a `decode` failure
### Caller
```python
store: Final = TransportResponsePollingStore(redis_cache.transport())
match await store.get(polling_id):
case RedisFailure():
return None
case None:
return None
case state:
return state
```
The caller keeps its own policy (here, fail open on any failure) and drops its timing, service hooks and health logging
### Tests
`tests/unit/redis/transport/test_redis_py.py` covers namespace applied once, TTL landing (`pttl`), `server` on a rejected command, `transport` on `server.connected = False`, `breaker_open` without touching the socket, the cluster gate being entered on a `RedisClusterCache`, and `eval` surviving `SCRIPT FLUSH`. Store tests go through the caller on `fake_redis_cache(server)` and seed with the old writer
## 2. Every remaining raw redis-py keyspace behind a store
Python only. The transport grows only the commands these keyspaces need (`incrby`, `expire`, `rpush`, `lpop`, `llen` as they come up), each with its fakeredis test. Before adding one, check whether the caller's sequence is really one `eval`
### Store per keyspace
One module each in `litellm/redis/stores/`
```text
mcp_single_use.py McpSingleUseStore claim, peek
pod_lock.py PodLockStore acquire, release
rate_limit.py RateLimitStore check_and_increment, decrement, read_windows
session_budget.py SessionBudgetStore add_spend, spend
iteration_count.py IterationCountStore increment
enqueued_tokens.py EnqueuedTokensStore add, total
login_throttle.py LoginThrottleStore record_failure, failures, clear
affinity.py AffinityStore pin, pinned
mcp_dcr.py McpDcrStore put_client, get_client
mcp_credential_refresh.py McpRefreshLockStore acquire, release
```
Operation names are a starting point, each PR names them for what its caller wants. The spend update buffer is not here, it reaches Redis through `RedisCache` list pipelines, which stay
### Example, `litellm/redis/stores/mcp_single_use.py`
```python
_CLAIM: Final = """
local count = redis.call("incrbyfloat", KEYS[1], ARGV[1])
if redis.call("ttl", KEYS[1]) < 0 then
redis.call("expire", KEYS[1], ARGV[2])
end
return count
"""
class McpSingleUseStore(Protocol):
async def claim(self, key: str, *, ttl: timedelta) -> int | RedisFailure: ...
async def peek(self, key: str) -> bool | RedisFailure: ...
@dataclass(frozen=True, slots=True)
class TransportMcpSingleUseStore:
transport: RedisTransport
async def claim(self, key: str, *, ttl: timedelta) -> int | RedisFailure:
reply: Final = await self.transport.eval(_CLAIM, (key,), (1, ttl // timedelta(seconds=1)))
return reply if isinstance(reply, RedisFailure) else _count(reply)
async def peek(self, key: str) -> bool | RedisFailure:
reply: Final = await self.transport.get(key)
return reply if isinstance(reply, RedisFailure) else reply is not None
```
`claim` is what `async_increment(key, 1, ttl=...)` does today, `INCRBYFLOAT` then `EXPIRE` only when the key has no TTL, in one script instead of a pipeline plus a conditional third round trip. The value stays an `INCRBYFLOAT` string, so `_count` parses a float and returns its integer, and a non-numeric reply is a `decode` failure. The caller keeps `"first"` as count 1, `"replayed"` otherwise, and fails closed on any `RedisFailure`. `peek` and `claim` move in the same commit, since they are one keyspace
### Example, `litellm/redis/stores/pod_lock.py`
```python
_RELEASE: Final = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
def lock_key(cronjob_id: str) -> str:
return f"cronjob_lock:{cronjob_id}"
@dataclass(frozen=True, slots=True)
class Acquired:
pass
@dataclass(frozen=True, slots=True)
class HeldBy:
owner: str
class PodLockStore(Protocol):
async def acquire(self, cronjob_id: str, owner: str, *, ttl: timedelta) -> Acquired | HeldBy | None | RedisFailure: ...
async def release(self, cronjob_id: str, owner: str) -> bool | RedisFailure: ...
@dataclass(frozen=True, slots=True)
class TransportPodLockStore:
transport: RedisTransport
async def acquire(self, cronjob_id: str, owner: str, *, ttl: timedelta) -> Acquired | HeldBy | None | RedisFailure:
set_reply: Final = await self.transport.set(lock_key(cronjob_id), _encode(owner), ttl=ttl, nx=True)
if isinstance(set_reply, RedisFailure):
return set_reply
if set_reply:
return Acquired()
held: Final = await self.transport.get(lock_key(cronjob_id))
if held is None or isinstance(held, RedisFailure):
return held
return _decode(held)
async def release(self, cronjob_id: str, owner: str) -> bool | RedisFailure:
reply: Final = await self.transport.eval(_RELEASE, (lock_key(cronjob_id),), (_encode(owner),))
return reply if isinstance(reply, RedisFailure) else reply == 1
def _encode(owner: str) -> str:
return json.dumps(owner)
def _decode(raw: bytes) -> HeldBy | RedisFailure: ...
```
The value is `json.dumps(pod_id)`, the bytes `async_set_cache` writes today, and the TTL stays whole seconds from the caller. `None` from `acquire` is the lock expiring between the `SET NX` and the `GET`, which the caller already treats as not acquired
The caller keeps the policy. `allow_reentrant` decides whether `HeldBy(self.pod_id)` counts as acquired, any `RedisFailure` is `False`, and the acquired and released events stay in `PodLockManager`. The GET plus DEL fallback after a failed script goes, as accepted in `migration-plan.md`, unless the pre-merge check finds a supported Redis setup with scripting disabled, in which case it stays in `PodLockManager`
### Lint gate
```toml
[lint.flake8-tidy-imports.banned-api]
"redis".msg = "reach Redis through litellm.redis (see .agents/skills/redis-access)"
```
Allowed in `litellm/redis/transport/redis_py.py`, `litellm/caching/redis_cache.py` and `litellm/caching/redis_cluster_cache.py` (they import `redis.exceptions` for the breaker and own the generic verbs), plus a per-file ignore for each remaining redis-py user listed in `migration-plan.md`, so the list only shrinks
## 3. Rust client parity
Rust only, nothing calls the new paths yet. This is 49189, and its PR body is the source of truth. The surface milestone 4 relies on
```rust
// crates/redis-client
impl RedisClient {
pub async fn connect(config: RedisConfig, events: Arc<dyn RedisEvents>) -> Result<Self, ConnectError>;
pub async fn get<T: FromRedisValue>(&self, key: &str) -> Result<Option<T>, Error>;
pub async fn set_nx(&self, key: &str, value: impl ToSingleRedisArg, ttl: Duration) -> Result<bool, Error>;
pub async fn eval<T: FromRedisValue>(&self, script: &Script, keys: &[&str], args: impl ToRedisArgs) -> Result<T, Error>;
pub fn breaker_state(&self) -> BreakerState;
}
pub enum Error { Transport(RedisError), Server(RedisError), Decode(RedisError), BreakerOpen }
// crates/redis-connect
pub struct RedisConfig { /* standalone, cluster, sentinel, tls, credential source, namespace, timeouts */ }
```
`Error` maps one to one onto `RedisFailure.kind`, and keys get the namespace inside the client exactly like the transport
## 4. Bridge and the first native store
### `litellm-rust/crates/redis-store/src/pod_lock.rs`
```rust
const RELEASE: &str = r#"
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"#;
pub fn lock_key(cronjob_id: &str) -> String {
format!("cronjob_lock:{cronjob_id}")
}
pub enum Acquire {
Acquired,
HeldBy(String),
Expired,
}
pub async fn acquire(client: &RedisClient, cronjob_id: &str, owner: &str, ttl: Duration) -> Result<Acquire, Error>;
pub async fn release(client: &RedisClient, cronjob_id: &str, owner: &str) -> Result<bool, Error>;
fn encode(owner: &str) -> String {
python_compat::json::dumps(&Value::Str(owner.into())).expect("a string always encodes")
}
```
Same keys, script text and codec as the Python store. The codec goes through `python-compat` so it matches `json.dumps` including ASCII escaping
### `litellm-rust/crates/python-bridge/src/redis/`
```rust
// client.rs, one per process, shared by every native store
#[pyclass(frozen)]
pub struct NativeRedisClient(Arc<RedisClient>);
#[pymethods]
impl NativeRedisClient {
#[staticmethod]
fn connect(settings: RedisSettings) -> PyResult<Self>;
}
// pod_lock.rs
#[pyclass(frozen)]
pub struct NativePodLockHandle(Arc<RedisClient>);
#[pymethods]
impl NativePodLockHandle {
#[new]
fn new(client: &NativeRedisClient) -> Self;
fn acquire<'py>(&self, py: Python<'py>, cronjob_id: String, owner: String, ttl_ms: u64) -> PyResult<Bound<'py, PyAny>>;
fn release<'py>(&self, py: Python<'py>, cronjob_id: String, owner: String) -> PyResult<Bound<'py, PyAny>>;
}
```
Each awaitable resolves to a generated frozen Pydantic result or a `RedisFailure`-shaped error value, never a raised exception for a Redis problem. The breaker lives in `RedisClient`, and its transitions reach `redis_breaker_state_counts()` through one host call in `litellm/rust_bridge/host/`
### `litellm/redis/stores/pod_lock.py`
```python
@dataclass(frozen=True, slots=True)
class NativePodLockStore:
native: NativePodLockHandle
events: RedisServiceEvents
async def acquire(self, cronjob_id: str, owner: str, *, ttl: timedelta) -> Acquired | HeldBy | None | RedisFailure:
return await self.events.observe("acquire", lambda: _acquire_reply(self.native.acquire(cronjob_id, owner, ttl_ms(ttl))))
async def release(self, cronjob_id: str, owner: str) -> bool | RedisFailure:
return await self.events.observe("release", lambda: _release_reply(self.native.release(cronjob_id, owner)))
def pod_lock_store(redis_cache: RedisCache, native: NativeRedisClient | None) -> PodLockStore:
if native is None:
return TransportPodLockStore(redis_cache.transport())
return NativePodLockStore(NativePodLockHandle(native), RedisServiceEvents(redis_cache.service_logger_obj))
```
`native` is `None` when the Rust extension is off or the Redis settings use something the Rust client does not support yet. That selection happens once at construction and is the only branch. `PodLockManager` takes a `PodLockStore` instead of a `RedisCache`
### Tests
`tests/integration/redis/test_pod_lock_cross_impl.py` runs on one real Redis from `litellm-redis-testing`, standalone and cluster
```python
@pytest.mark.parametrize(("holder", "contender"), [(transport, native), (native, transport)])
async def test_lock_taken_by_one_stack_is_seen_and_released_only_by_its_owner(holder, contender): ...
```
It also checks that the TTL lands, that a Rust pod cannot release a Python pod's lock and the reverse, and that the owner releases across stacks. Rust side tests for `redis-store` run against the same Redis
### Caller switch
A separate commit or PR. The proxy builds `NativeRedisClient` once at startup and passes it to `pod_lock_store`. Proof is a live proxy pair, one pod on each stack, contending for the same cron lock

76
.agents/skills/redis-access/references/transport.md
View file

@ -1,76 +0,0 @@
# RedisTransport
How Python code reaches Redis. Read this before adding a command, or moving raw redis-py access off `RedisCache`
## Why it exists
Today commands reach Redis three ways
- Cache methods on `RedisCache` and `RedisBatch` (get, set, incr, lists, pipelines), which apply the namespace and the circuit breaker
- Raw access, where callers take `init_async_client()`, `init_pubsub_client()`, `async_register_script()` or `.redis_client` and run commands, Lua and pub/sub themselves
- Scripts from `async_register_script` keep the namespace and breaker
- Raw clients skip both, so every caller namespaces keys and channels by hand, and some forget
- Vector search (`RedisSemanticCache` through RedisVL, `ValkeySemanticCache` and the Valkey vector store through `Redis.from_url`), which builds its own clients on separate connections
`RedisTransport` replaces the raw path with one command-shaped seam, so the backend behind it can later swap from redis-py to `litellm-rust/crates/redis-client` without touching callers
## The contract
Defined in `litellm/redis/transport/contract.py`
- Commands are frozen dataclasses, `Get(key)`, `Set(key, value, ttl, nx=False)` and `Delete(key)`. `Arg` is `str | bytes | int | float`
- The transport has one method, `execute(command)`, overloaded per command for its reply type (`bytes | None`, `bool`, `int`)
- Every command returns its value or a `RedisFailure`. Nothing raises for a Redis problem
- `RedisFailure.kind`
- `transport` is a timeout or lost connection. The command may still have applied, so a write that fails this way has an unknown outcome. `timeout=True` marks timeouts
- `server` means Redis answered with an error (WRONGTYPE). The connection is fine and the breaker does not count it
- `decode` means the reply was not the shape the method or store expected
- `breaker_open` means the breaker refused the call without touching the socket
- The namespace is applied once, inside the transport. A key that already starts with `{namespace}:` is left alone, matching `RedisCache.check_and_fix_namespace` and `crates/redis-client`. Callers and stores pass logical keys
- Every write takes its TTL from the caller. `ttl=None` means no expiry, there is no default
- Replies are undecoded bytes. Decoding belongs to whoever owns the keyspace
These rules are shared with `crates/redis-client` (its `AGENTS.md` has the Rust side). If you change one, change both, since Python and Rust pods share keys in one Redis
## The redis-py backend
- `litellm/redis/transport/redis_py.py` holds `RedisPyTransport`, the only module that touches redis-py's client
- `RedisCache.transport()` builds one over that cache's own async client, namespace and circuit breaker. No new connection, no second breaker
- The guard is `BreakerGuard` in `redis_cache.py`. It runs the call under the cache's breaker and maps exceptions with `_redis_failure`
- `ResponseError` / `DataError` that are not health failures become `server`
- health failures and other `RedisError`s become `transport`
- anything else re-raises, since it is a bug, not a Redis failure
- `BreakerGuard` wraps `ReportingGuard`, which owns observability for every command
- one REDIS service success or failure event per command, with the command name as `call_type`, sent to the cache's `service_logger_obj`. A `breaker_open` refusal sends none
- the event and the log line carry the caller's `with_service_target`, so callers label their commands and emit nothing themselves
- one WARNING per failure through the timeout log throttle, `breaker_open` at DEBUG
- Callers don't log Redis health. Log only what a failure means for the feature, and only when it adds something the transport line doesn't say
## Adding a command
- Add a frozen dataclass for the command to `contract.py`, add it to `Command`, and add an `execute` overload for its reply type. Keep it command-shaped, never redis-py shaped (no `**kwargs`, no client options), so the native client fits behind it unchanged
- Add a `match` arm in every backend's `execute`, and add the redis-py method to the `AsyncRedisCommands` Protocol there and `_AsyncRedisCommands` in `redis_cache.py`
- Namespace every key argument with `_key`, take TTLs as `timedelta`, send them as milliseconds
- Validate the reply shape and return `decode` for anything unexpected
- Test it in `tests/unit/redis/transport/test_redis_py.py` against fakeredis: the namespace is applied once, TTL lands, a rejected command is `server`, an outage is `transport`. Service events and logging come from the guard, so a new command needs no test of its own for them
- Multi-key commands must keep their keys in one slot, or they break on cluster depending on slot ownership
## Moving raw access onto the transport
- Replace the raw client or script handle with transport calls, and delete hand-written namespacing and script registration
- Keep the caller's outcome per failure. If it failed closed on any exception, it fails closed on any `RedisFailure`
- Don't treat a `transport` failure on a write as "not applied"
## Not on the transport yet
- Pub/sub (config and auth-cache invalidation), admin commands, pipelines and `RedisBatch` stay on `RedisCache` until the transport grows them
- Vector and semantic search keep their own clients. The plan gives them native index handles on `crates/redis-client`, not `RedisTransport`
## Where it's going
- A `NativeRedisTransport` (a PyO3 adapter over one `RedisClient` per resolved target) goes behind the same Protocol and takes the same `Guard`. The backend is picked at construction, with redis-py kept for config the native client doesn't support yet (Sentinel, cloud auth, TLS cert options)
- Settings for both stacks resolve through `crates/config` (`RedisSettings.resolve`), so Python and Rust connect to the same target with the same namespace
- During the swap the Python `RedisCircuitBreaker` still wraps the transport. The breaker and the service events have to be in Rust before any store runs natively, because native stores call `crates/redis-client` directly
- Once stores and admin/pub-sub have replacements, redis-py and redisvl are dropped or become optional extras
Don't build any of this unless asked

11
litellm/_service_logger.py
View file

@ -111,6 +111,9 @@ class ServiceLogging(CustomLogger):
return _as_otel_logger(open_telemetry_logger)
return None
def has_sinks(self) -> bool:
return bool(litellm.service_callback)
@staticmethod
def _sync_dispatch_loop() -> asyncio.AbstractEventLoop | None:
"""The event loop a blocking caller can dispatch on, or ``None`` if it has none."""
@ -121,7 +124,7 @@ class ServiceLogging(CustomLogger):
return None if loop.is_closed() else loop
@staticmethod
async def _emit_guarded(hook: Callable[[], Coroutine[object, object, None]]) -> None:
async def emit_guarded(hook: Callable[[], Coroutine[object, object, None]]) -> None:
"""Emit one service event, absorbing anything the callbacks raise.
Monitoring must not break the call it monitors. Sync callers are the ones that
@ -144,11 +147,11 @@ class ServiceLogging(CustomLogger):
loop: Final = ServiceLogging._sync_dispatch_loop()
try:
if loop is None:
asyncio.run(ServiceLogging._emit_guarded(hook))
asyncio.run(ServiceLogging.emit_guarded(hook))
elif loop.is_running():
loop.create_task(ServiceLogging._emit_guarded(hook))
loop.create_task(ServiceLogging.emit_guarded(hook))
else:
loop.run_until_complete(ServiceLogging._emit_guarded(hook))
loop.run_until_complete(ServiceLogging.emit_guarded(hook))
except Exception as e:
verbose_logger.exception("Error dispatching service event - %s", e)

99
litellm/caching/redis_cache.py
View file

@ -30,7 +30,7 @@ from typing import TYPE_CHECKING, Any, Final, Protocol, TypeVar, cast
from pydantic import TypeAdapter
import litellm
from litellm._internal_context import current_service_caller, current_service_target
from litellm._internal_context import current_service_caller
from litellm._logging import print_verbose, verbose_logger
from litellm.constants import (
DEFAULT_REDIS_MAJOR_VERSION,
@ -84,6 +84,8 @@ class _AsyncRedisCommands(Protocol):
def get(self, name: str) -> Awaitable[object]: ...
def mget(self, keys: Sequence[str]) -> Awaitable[object]: ...
def set(self, name: str, value: str | bytes | float, *, nx: bool, px: int | None) -> Awaitable[object]: ...
def delete(self, *names: str) -> Awaitable[int]: ...
@ -100,6 +102,8 @@ class _AsyncRedisCommands(Protocol):
def eval(self, script: str, numkeys: int, *keys_and_args: str | bytes | float) -> Awaitable[object]: ...
def evalsha(self, sha: str, numkeys: int, *keys_and_args: str | bytes | float) -> Awaitable[object]: ...
_GENERIC_CALLER_MODULES: Final = frozenset(
{
@ -108,7 +112,6 @@ _GENERIC_CALLER_MODULES: Final = frozenset(
"litellm.caching.dual_cache",
"litellm.caching.caching",
"litellm.rust_bridge.execution",
"litellm.redis.transport.redis_py",
"contextlib",
}
)
@ -157,7 +160,11 @@ def _decoded_counts(values: Sequence[bytes | str | None]) -> tuple[int | None, .
def _is_generic_caller_frame(frame: FrameType) -> bool:
module: Final = frame.f_globals.get("__name__")
return module in _GENERIC_CALLER_MODULES or frame.f_code.co_name in _GENERIC_CALLER_FRAME_NAMES
return (
module in _GENERIC_CALLER_MODULES
or (isinstance(module, str) and module.startswith("litellm.redis."))
or frame.f_code.co_name in _GENERIC_CALLER_FRAME_NAMES
)
def _ends_call_stack(frame: FrameType) -> bool:
@ -610,85 +617,10 @@ async def _run_under_circuit_breaker(
return result
def _redis_failure(exc: Exception) -> RedisFailure | None:
from redis.exceptions import DataError, RedisError, ResponseError
def _is_redis_error(exc: BaseException) -> bool:
from redis.exceptions import RedisError
if isinstance(exc, (ResponseError, DataError)) and not _is_redis_health_failure(exc):
return RedisFailure(kind="server", message=str(exc))
if _is_redis_health_failure(exc) or isinstance(exc, RedisError):
return RedisFailure(kind="transport", message=str(exc), timeout=is_redis_timeout_failure(exc))
return None
class _Admit(Protocol):
def __call__(self, call: Callable[[], Awaitable[_RedisCallResult]], /) -> Awaitable[_RedisCallResult]: ...
def _admit_directly(call: Callable[[], Awaitable[_RedisCallResult]]) -> Awaitable[_RedisCallResult]:
return call()
@dataclass(frozen=True, slots=True)
class ReportingGuard:
events: "ServiceLogging"
async def __call__(self, name: str, call: Callable[[], Awaitable[_RedisCallResult]], /) -> _RedisCallResult:
caller: Final = _get_call_stack_info()
start_time: Final = time.time()
result: Final = await call()
end_time: Final = time.time()
if not isinstance(result, RedisFailure):
asyncio.create_task(
self.events.async_service_success_hook(
service=ServiceTypes.REDIS,
call_type=name,
duration=end_time - start_time,
start_time=start_time,
end_time=end_time,
caller=caller,
)
)
return result
target: Final = current_service_target()
_log_redis_failure_value(
verbose_logger, logging.WARNING, f"Redis {name} failed" + (f" for {target}" if target else ""), result
)
if result.kind != "breaker_open":
asyncio.create_task(
self.events.async_service_failure_hook(
service=ServiceTypes.REDIS,
duration=end_time - start_time,
error=result.message,
call_type=name,
start_time=start_time,
end_time=end_time,
caller=caller,
)
)
return result
@dataclass(frozen=True, slots=True)
class BreakerGuard:
breaker: RedisCircuitBreaker
report: ReportingGuard
admit: _Admit = _admit_directly
async def __call__(
self, name: str, call: Callable[[], Awaitable[_RedisCallResult]], /
) -> _RedisCallResult | RedisFailure:
return await self.report(name, lambda: self._run(name, call))
async def _run(self, name: str, call: Callable[[], Awaitable[_RedisCallResult]]) -> _RedisCallResult | RedisFailure:
try:
return await _run_under_circuit_breaker(self.breaker, name, lambda: self.admit(call))
except RedisCircuitBreakerOpenError as e:
return RedisFailure(kind="breaker_open", message=str(e))
except Exception as e:
failure: Final = _redis_failure(e)
if failure is None:
raise
return failure
return isinstance(exc, RedisError)
def _run_under_circuit_breaker_sync(
@ -937,12 +869,15 @@ class RedisCache(BaseCache):
return self.init_async_client()
def transport(self) -> RedisTransport:
from litellm.redis.transport.events import RedisServiceEvents
from litellm.redis.transport.redis_py import RedisPyTransport
return RedisPyTransport(
client=self._async_commands,
namespace=self.namespace,
guard=BreakerGuard(self._circuit_breaker, ReportingGuard(self.service_logger_obj), self._admit_async_call),
breaker=self._circuit_breaker,
admit=self._admit_async_call,
events=RedisServiceEvents(self.service_logger_obj),
)
def check_and_fix_namespace(self, key: str) -> str:

75
litellm/proxy/response_polling/polling_handler.py
View file

@ -2,9 +2,8 @@
Response Polling Handler for Background Responses with Cache
"""
import json
from collections.abc import Mapping
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from typing import Final
from pydantic import TypeAdapter
@ -13,22 +12,26 @@ from litellm._internal_context import with_service_target
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid4
from litellm.caching.redis_cache import RedisCache
from litellm.redis.stores import response_polling
from litellm.redis.stores.response_polling import ResponsePollingStore, TransportResponsePollingStore
from litellm.redis.transport.contract import RedisFailure
from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesAPIStatus
_RESPONSE_POLLING_TARGET: Final = "response_polling"
_JSON_OBJECT: Final = TypeAdapter(dict[str, object])
_JSON_LIST: Final = TypeAdapter(list[object])
class ResponsePollingHandler:
"""Handles polling-based responses with Redis cache"""
CACHE_KEY_PREFIX = "litellm:polling:response:"
CACHE_KEY_PREFIX = response_polling.KEY_PREFIX
POLLING_ID_PREFIX = "litellm_poll_" # Clear prefix to identify polling IDs
def __init__(self, redis_cache: RedisCache | None = None, ttl: int = 3600):
self.redis_cache = redis_cache
self.ttl = ttl # Time-to-live for cache entries (default: 1 hour)
self.ttl = ttl
self._store: Final[ResponsePollingStore | None] = (
TransportResponsePollingStore(redis_cache.transport()) if redis_cache is not None else None
)
@classmethod
def generate_polling_id(cls) -> str:
@ -43,7 +46,10 @@ class ResponsePollingHandler:
@classmethod
def get_cache_key(cls, polling_id: str) -> str:
"""Get Redis cache key for a polling ID"""
return f"{cls.CACHE_KEY_PREFIX}{polling_id}"
return response_polling.state_key(polling_id)
def _ttl(self) -> timedelta:
return timedelta(seconds=self.ttl)
@with_service_target(_RESPONSE_POLLING_TARGET)
async def create_initial_state(
@ -79,16 +85,10 @@ class ResponsePollingHandler:
}
)
cache_key: Final = self.get_cache_key(polling_id)
if self.redis_cache:
# Store ResponsesAPIResponse directly in Redis
await self.redis_cache.async_set_cache(
key=cache_key,
value=response.model_dump_json(), # Pydantic v2 method
ttl=self.ttl,
)
verbose_proxy_logger.debug("Created initial polling state for %s with TTL=%ss", polling_id, self.ttl)
if self._store is not None:
written: Final = await self._store.put(polling_id, response.model_dump(mode="json"), ttl=self._ttl())
if written is None:
verbose_proxy_logger.debug("Created initial polling state for %s with TTL=%ss", polling_id, self.ttl)
return response
@ -145,19 +145,17 @@ class ResponsePollingHandler:
user: User identifier
store: Whether to store the response
"""
if not self.redis_cache:
if self._store is None:
return
cache_key: Final = self.get_cache_key(polling_id)
# Get current state
cached_state: Final = await self.redis_cache.async_get_cache(cache_key)
if not cached_state:
cached_state: Final = await self._store.get(polling_id)
if isinstance(cached_state, RedisFailure):
return
if cached_state is None:
verbose_proxy_logger.warning("No cached state found for polling_id: %s", polling_id)
return
# Parse existing ResponsesAPIResponse from cache
state: Final = _JSON_OBJECT.validate_python(json.loads(cached_state))
state: Final[dict[str, object]] = dict(cached_state)
# Update status (using OpenAI native status values)
if status:
@ -212,12 +210,9 @@ class ResponsePollingHandler:
if store is not None:
state["store"] = store
# Update cache with configured TTL
await self.redis_cache.async_set_cache(
key=cache_key,
value=json.dumps(state),
ttl=self.ttl,
)
written: Final = await self._store.put(polling_id, state, ttl=self._ttl())
if written is not None:
return
output_count: Final = len(_JSON_LIST.validate_python(state.get("output", [])))
verbose_proxy_logger.debug(
@ -227,16 +222,11 @@ class ResponsePollingHandler:
@with_service_target(_RESPONSE_POLLING_TARGET)
async def get_state(self, polling_id: str) -> Mapping[str, object] | None:
"""Get current polling state from Redis"""
if not self.redis_cache:
if self._store is None:
return None
cache_key: Final = self.get_cache_key(polling_id)
cached_state: Final = await self.redis_cache.async_get_cache(cache_key)
if cached_state:
return _JSON_OBJECT.validate_python(json.loads(cached_state))
return None
cached_state: Final = await self._store.get(polling_id)
return None if isinstance(cached_state, RedisFailure) else cached_state
async def cancel_polling(self, polling_id: str) -> bool:
"""
@ -253,13 +243,10 @@ class ResponsePollingHandler:
@with_service_target(_RESPONSE_POLLING_TARGET)
async def delete_polling(self, polling_id: str) -> bool:
"""Delete a polling request from cache"""
if not self.redis_cache:
if self._store is None:
return False
cache_key: Final = self.get_cache_key(polling_id)
# Use RedisCache's async_delete_cache method which handles Redis/RedisCluster
await self.redis_cache.async_delete_cache(cache_key)
return True
return await self._store.delete(polling_id) is None
def should_use_polling_for_request(

15
litellm/redis/AGENTS.md
View file

@ -1,11 +1,12 @@
For adding Redis access or moving raw Redis access here, follow [.agents/skills/redis-access/SKILL.md](../../.agents/skills/redis-access/SKILL.md)
- `transport/contract.py` is the command-shaped seam. Keep it one command dataclass per Redis command the Python code uses, run through `execute`, never redis-py shaped, so `litellm-rust/crates/redis-client` can sit behind it unchanged
- Rules shared with `crates/redis-client`, so the two implementations agree
- The namespace is applied here, once. A key that already starts with `{namespace}:` is left alone
- `transport/contract.py` is the Python seam to redis-py. `RedisTransport` is plain methods, one per Redis command the Python code uses, command shaped and never redis-py shaped. It is not a seam for the Rust client, native stores sit behind a store `Protocol` instead
- Rules shared with `crates/redis-client`, so Python and Rust pods agree on one Redis
- The namespace is applied here, once, including Lua `KEYS`. A key that already starts with `{namespace}:` is left alone
- Every write takes its TTL from the caller, `None` means no expiry, there is no default
- Replies are undecoded bytes
- Replies are undecoded bytes, `mget` and `eval` arrays are tuples
- Failures are values. `transport` covers timeouts and connection loss, where the command may still have applied. `server` means Redis answered with an error and the connection is fine
- `transport/redis_py.py` is the only module that touches redis-py's client. The breaker and the exception mapping come from `RedisCache` through the injected guard
- Every backend runs each command through a `Guard` that emits one REDIS service event and logs Redis health failures, so swapping backends never drops metrics. Callers don't log Redis health
- `litellm/caching` imports `litellm/redis`, never the reverse
- `transport/redis_py.py` is the only module here that imports redis-py. `transport/breaker.py` reuses `RedisCache`'s breaker admission and health classification, it never reimplements them
- `transport/events.py` is the only place REDIS service events for transport traffic are emitted. Stores and callers emit nothing and do not log Redis health
- `stores/` holds one module per keyspace, and a keyspace that has a store is only reached through it. A store is a `Protocol` of domain operations plus a frozen dataclass over `RedisTransport`. It owns its keys, value codec, expiry, Lua and decoding, and callers never see a Redis command
- Keys, value bytes and TTL units in Redis are a wire format shared with older pods and Rust. Never change them in place

0
litellm/redis/stores/__init__.py Normal file
View file

48
litellm/redis/stores/response_polling.py Normal file
View file

@ -0,0 +1,48 @@
import json
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import timedelta
from typing import Final, Protocol
from pydantic import JsonValue, TypeAdapter, ValidationError
from litellm.redis.transport.contract import RedisFailure, RedisTransport
KEY_PREFIX: Final = "litellm:polling:response:"
_DOCUMENT: Final = TypeAdapter(str)
_STATE: Final = TypeAdapter(dict[str, JsonValue])
def state_key(polling_id: str) -> str:
return f"{KEY_PREFIX}{polling_id}"
class ResponsePollingStore(Protocol):
async def put(self, polling_id: str, state: Mapping[str, object], *, ttl: timedelta) -> None | RedisFailure: ...
async def get(self, polling_id: str) -> Mapping[str, JsonValue] | None | RedisFailure: ...
async def delete(self, polling_id: str) -> None | RedisFailure: ...
@dataclass(frozen=True, slots=True)
class TransportResponsePollingStore:
transport: RedisTransport
async def put(self, polling_id: str, state: Mapping[str, object], *, ttl: timedelta) -> None | RedisFailure:
written: Final = await self.transport.set(state_key(polling_id), json.dumps(json.dumps(state)), ttl=ttl)
return written if isinstance(written, RedisFailure) else None
async def get(self, polling_id: str) -> Mapping[str, JsonValue] | None | RedisFailure:
raw: Final = await self.transport.get(state_key(polling_id))
if raw is None or isinstance(raw, RedisFailure):
return raw
try:
return _STATE.validate_json(_DOCUMENT.validate_json(raw))
except ValidationError as e:
return RedisFailure(kind="decode", message=f"polling state is not a JSON object: {e}")
async def delete(self, polling_id: str) -> None | RedisFailure:
deleted: Final = await self.transport.delete(state_key(polling_id))
return deleted if isinstance(deleted, RedisFailure) else None

36
litellm/redis/transport/breaker.py Normal file
View file

@ -0,0 +1,36 @@
from collections.abc import Awaitable, Callable
from typing import Final, TypeVar
from litellm.caching.redis_cache import (
RedisCircuitBreaker,
RedisCircuitBreakerOpenError,
_is_redis_error, # pyright: ignore[reportPrivateUsage] # same exception classes as RedisCache
_is_redis_health_failure, # pyright: ignore[reportPrivateUsage] # same health classification as RedisCache
_run_under_circuit_breaker, # pyright: ignore[reportPrivateUsage] # same admission and health signal as RedisCache
is_redis_timeout_failure,
)
from litellm.redis.transport.contract import Admit, RedisFailure
_T: Final = TypeVar("_T")
async def run_under_breaker(
breaker: RedisCircuitBreaker, name: str, admit: Admit, call: Callable[[], Awaitable[_T]]
) -> _T | RedisFailure:
try:
return await _run_under_circuit_breaker(breaker, name, lambda: admit(call))
except RedisCircuitBreakerOpenError as e:
return RedisFailure(kind="breaker_open", message=str(e))
except Exception as e:
failure: Final = _redis_failure(e)
if failure is None:
raise
return failure
def _redis_failure(exc: BaseException) -> RedisFailure | None:
if _is_redis_health_failure(exc):
return RedisFailure(kind="transport", message=str(exc), timeout=is_redis_timeout_failure(exc))
if _is_redis_error(exc):
return RedisFailure(kind="server", message=str(exc))
return None

47
litellm/redis/transport/contract.py
View file

@ -1,14 +1,19 @@
import math
from collections.abc import Awaitable, Callable
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass
from datetime import timedelta
from typing import Final, Literal, Protocol, TypeAlias, TypeVar, overload
from typing import Final, Literal, Protocol, TypeAlias, TypeVar
Arg: TypeAlias = str | bytes | int | float
EvalReply: TypeAlias = int | bytes | None | tuple["EvalReply", ...]
_T: Final = TypeVar("_T")
class Admit(Protocol):
def __call__(self, call: Callable[[], Awaitable[_T]], /) -> Awaitable[_T]: ...
@dataclass(frozen=True, slots=True)
class RedisFailure:
kind: Literal["transport", "server", "decode", "breaker_open"]
@ -16,41 +21,17 @@ class RedisFailure:
timeout: bool = False
@dataclass(frozen=True, slots=True)
class Get:
key: str
@dataclass(frozen=True, slots=True)
class Set:
key: str
value: Arg
ttl: timedelta | None
nx: bool = False
@dataclass(frozen=True, slots=True)
class Delete:
key: str
Command: TypeAlias = Get | Set | Delete
class Guard(Protocol):
def __call__(self, name: str, call: Callable[[], Awaitable[_T]], /) -> Awaitable[_T | RedisFailure]: ...
def ttl_ms(ttl: timedelta) -> int:
return max(1, math.ceil(ttl / timedelta(milliseconds=1)))
class RedisTransport(Protocol):
@overload
async def execute(self, command: Get) -> bytes | None | RedisFailure: ...
async def get(self, key: str) -> bytes | None | RedisFailure: ...
@overload
async def execute(self, command: Set) -> bool | RedisFailure: ...
async def mget(self, keys: Sequence[str]) -> tuple[bytes | None, ...] | RedisFailure: ...
@overload
async def execute(self, command: Delete) -> int | RedisFailure: ...
async def set(self, key: str, value: Arg, *, ttl: timedelta | None, nx: bool = False) -> bool | RedisFailure: ...
async def delete(self, *keys: str) -> int | RedisFailure: ...
async def eval(self, script: str, keys: Sequence[str], args: Sequence[Arg]) -> EvalReply | RedisFailure: ...

71
litellm/redis/transport/events.py Normal file
View file

@ -0,0 +1,71 @@
import logging
import time
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Final, TypeVar
from litellm._internal_context import current_service_target
from litellm._logging import verbose_logger
from litellm._service_logger import ServiceLogging
from litellm.caching.redis_cache import (
_get_call_stack_info, # pyright: ignore[reportPrivateUsage] # same caller attribution as RedisCache events
_log_redis_failure_value, # pyright: ignore[reportPrivateUsage] # same timeout throttle as RedisCache logging
)
from litellm.redis.transport.contract import RedisFailure
from litellm.types.services import ServiceTypes
_T: Final = TypeVar("_T")
@dataclass(frozen=True, slots=True)
class RedisServiceEvents:
service_logging: ServiceLogging
async def observe(self, call_type: str, call: Callable[[], Awaitable[_T | RedisFailure]]) -> _T | RedisFailure:
if not self.service_logging.has_sinks():
result: Final = await call()
_log_failure(call_type, result)
return result
caller: Final = _get_call_stack_info()
start_time: Final = time.time()
observed: Final = await call()
end_time: Final = time.time()
_log_failure(call_type, observed)
await self._emit(call_type, observed, caller=caller, start_time=start_time, end_time=end_time)
return observed
async def _emit(self, call_type: str, result: object, *, caller: str, start_time: float, end_time: float) -> None:
if not isinstance(result, RedisFailure):
await ServiceLogging.emit_guarded(
lambda: self.service_logging.async_service_success_hook(
service=ServiceTypes.REDIS,
call_type=call_type,
duration=end_time - start_time,
start_time=start_time,
end_time=end_time,
caller=caller,
)
)
return
if result.kind == "breaker_open":
return
await ServiceLogging.emit_guarded(
lambda: self.service_logging.async_service_failure_hook(
service=ServiceTypes.REDIS,
duration=end_time - start_time,
error=result.message,
call_type=call_type,
start_time=start_time,
end_time=end_time,
caller=caller,
)
)
def _log_failure(call_type: str, result: object) -> None:
if not isinstance(result, RedisFailure):
return
target: Final = current_service_target()
_log_redis_failure_value(
verbose_logger, logging.WARNING, f"Redis {call_type} failed" + (f" for {target}" if target else ""), result
)

149
litellm/redis/transport/redis_py.py
View file

@ -1,59 +1,144 @@
from collections.abc import Awaitable, Callable
import hashlib
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass
from typing import Final, Protocol, assert_never, overload
from datetime import timedelta
from typing import Final, Protocol, TypeVar
from litellm.redis.transport.contract import Arg, Command, Delete, Get, Guard, RedisFailure, Set, ttl_ms
from pydantic import TypeAdapter, ValidationError
from redis.exceptions import NoScriptError
from litellm.caching.redis_cache import RedisCircuitBreaker
from litellm.redis.transport.breaker import run_under_breaker
from litellm.redis.transport.contract import Admit, Arg, EvalReply, RedisFailure, ttl_ms
from litellm.redis.transport.events import RedisServiceEvents
_T: Final = TypeVar("_T")
_ARRAY: Final = TypeAdapter(tuple[object, ...])
class AsyncRedisCommands(Protocol):
def get(self, name: str) -> Awaitable[object]: ...
def mget(self, keys: Sequence[str]) -> Awaitable[object]: ...
def set(self, name: str, value: Arg, *, nx: bool, px: int | None) -> Awaitable[object]: ...
def delete(self, *names: str) -> Awaitable[int]: ...
def eval(self, script: str, numkeys: int, *keys_and_args: Arg) -> Awaitable[object]: ...
def _decode_failure(value: object) -> RedisFailure:
return RedisFailure(kind="decode", message=f"unexpected Redis reply of type {type(value).__name__}")
def _bytes_reply(reply: object) -> bytes | None | RedisFailure:
if reply is None or isinstance(reply, (bytes, RedisFailure)):
return reply
if isinstance(reply, str):
return reply.encode()
return _decode_failure(reply)
def evalsha(self, sha: str, numkeys: int, *keys_and_args: Arg) -> Awaitable[object]: ...
@dataclass(frozen=True, slots=True)
class RedisPyTransport:
client: Callable[[], AsyncRedisCommands]
namespace: str | None
guard: Guard
breaker: RedisCircuitBreaker
admit: Admit
events: RedisServiceEvents
async def get(self, key: str) -> bytes | None | RedisFailure:
return _bytes_reply(await self._run("get", lambda: self.client().get(self._key(key))))
async def mget(self, keys: Sequence[str]) -> tuple[bytes | None, ...] | RedisFailure:
namespaced: Final = tuple(self._key(k) for k in keys)
return _mget_reply(await self._run("mget", lambda: self.client().mget(namespaced)), len(namespaced))
async def set(self, key: str, value: Arg, *, ttl: timedelta | None, nx: bool = False) -> bool | RedisFailure:
px: Final = None if ttl is None else ttl_ms(ttl)
return _bool_reply(await self._run("set", lambda: self.client().set(self._key(key), value, px=px, nx=nx)))
async def delete(self, *keys: str) -> int | RedisFailure:
namespaced: Final = tuple(self._key(k) for k in keys)
return _int_reply(await self._run("delete", lambda: self.client().delete(*namespaced)))
async def eval(self, script: str, keys: Sequence[str], args: Sequence[Arg]) -> EvalReply | RedisFailure:
sha: Final = hashlib.sha1(script.encode(), usedforsecurity=False).hexdigest()
namespaced: Final = tuple(self._key(k) for k in keys)
return _eval_reply(
await self._run("eval", lambda: _evalsha_or_eval(self.client(), script, sha, namespaced, args))
)
def _key(self, key: str) -> str:
if not self.namespace or key.startswith(f"{self.namespace}:"):
return key
return f"{self.namespace}:{key}"
@overload
async def execute(self, command: Get) -> bytes | None | RedisFailure: ...
async def _run(self, name: str, call: Callable[[], Awaitable[_T]]) -> _T | RedisFailure:
return await self.events.observe(name, lambda: run_under_breaker(self.breaker, name, self.admit, call))
@overload
async def execute(self, command: Set) -> bool | RedisFailure: ...
@overload
async def execute(self, command: Delete) -> int | RedisFailure: ...
async def _evalsha_or_eval(
client: AsyncRedisCommands, script: str, sha: str, keys: Sequence[str], args: Sequence[Arg]
) -> object:
try:
return await client.evalsha(sha, len(keys), *keys, *args)
except NoScriptError:
return await client.eval(script, len(keys), *keys, *args)
async def execute(self, command: Command) -> bytes | None | bool | int | RedisFailure:
match command:
case Get(key=key):
return _bytes_reply(await self.guard("get", lambda: self.client().get(self._key(key))))
case Set(key=key, value=value, ttl=ttl, nx=nx):
px: Final = None if ttl is None else ttl_ms(ttl)
written: Final = await self.guard("set", lambda: self.client().set(self._key(key), value, nx=nx, px=px))
return written if isinstance(written, RedisFailure) else written is not None and written is not False
case Delete(key=key):
return await self.guard("delete", lambda: self.client().delete(self._key(key)))
case _:
assert_never(command)
def _decode_failure(reply: object) -> RedisFailure:
return RedisFailure(kind="decode", message=f"unexpected Redis reply of type {type(reply).__name__}")
def _optional_bytes(reply: object) -> bytes | None | RedisFailure:
if reply is None or isinstance(reply, bytes):
return reply
if isinstance(reply, str):
return reply.encode()
return _decode_failure(reply)
def _bytes_reply(reply: object) -> bytes | None | RedisFailure:
return reply if isinstance(reply, RedisFailure) else _optional_bytes(reply)
def _array(reply: object) -> tuple[object, ...] | None:
if not isinstance(reply, list):
return None
try:
return _ARRAY.validate_python(reply)
except ValidationError:
return None
def _mget_reply(reply: object, expected: int) -> tuple[bytes | None, ...] | RedisFailure:
if isinstance(reply, RedisFailure):
return reply
array: Final = _array(reply)
if array is None or len(array) != expected:
return _decode_failure(reply)
values: Final = tuple(_optional_bytes(value) for value in array)
failure: Final = next((value for value in values if isinstance(value, RedisFailure)), None)
return failure if failure is not None else tuple(value for value in values if not isinstance(value, RedisFailure))
def _bool_reply(reply: object) -> bool | RedisFailure:
if isinstance(reply, RedisFailure):
return reply
return reply is not None and reply is not False
def _int_reply(reply: object) -> int | RedisFailure:
if isinstance(reply, RedisFailure):
return reply
return reply if isinstance(reply, int) and not isinstance(reply, bool) else _decode_failure(reply)
def _eval_reply(reply: object) -> EvalReply | RedisFailure:
if isinstance(reply, RedisFailure) or reply is None or isinstance(reply, bytes):
return reply
if isinstance(reply, bool):
return _decode_failure(reply)
if isinstance(reply, int):
return reply
if isinstance(reply, str):
return reply.encode()
array: Final = _array(reply)
if array is None:
return _decode_failure(reply)
items: Final = tuple(_eval_reply(item) for item in array)
failure: Final = next((item for item in items if isinstance(item, RedisFailure)), None)
return failure if failure is not None else tuple(item for item in items if not isinstance(item, RedisFailure))

2
pyproject.toml
View file

@ -233,7 +233,7 @@ dev = [
"opentelemetry-instrumentation-fastapi==0.66b1",
"langfuse>=4.7,<5.0",
"fastapi-offline==1.7.6",
"fakeredis==2.34.1",
"fakeredis[lua]==2.34.1",
"pytest-rerunfailures==15.1",
"pytest-cov==5.0.0",
"parameterized==0.9.0",

22
tests/_support/fake_redis.py
View file

@ -1,15 +1,20 @@
from collections.abc import Awaitable, Callable
from typing import TypeVar
import fakeredis
import litellm
from litellm._service_logger import ServiceLogging
from litellm.caching.redis_cache import (
BreakerGuard,
RedisCache,
RedisCircuitBreaker,
ReportingGuard,
)
from litellm.caching.redis_cache import RedisCache, RedisCircuitBreaker
from litellm.redis.transport.events import RedisServiceEvents
from litellm.redis.transport.redis_py import RedisPyTransport
_T = TypeVar("_T")
def _admit_directly(call: Callable[[], Awaitable[_T]]) -> Awaitable[_T]:
return call()
def fake_redis_cache(
server: fakeredis.FakeServer, namespace: str | None = None, service_logger: ServiceLogging | None = None
@ -35,9 +40,10 @@ def fake_redis_transport(
events: ServiceLogging | None = None,
namespace: str | None = None,
) -> RedisPyTransport:
breaker = RedisCircuitBreaker(failure_threshold=failure_threshold, recovery_timeout=3600)
return RedisPyTransport(
client=lambda: client,
namespace=namespace,
guard=BreakerGuard(breaker, ReportingGuard(events or ServiceLogging())),
breaker=RedisCircuitBreaker(failure_threshold=failure_threshold, recovery_timeout=3600),
admit=_admit_directly,
events=RedisServiceEvents(events or ServiceLogging()),
)

309
tests/unit/proxy/response_polling/test_response_polling_handler.py
View file

@ -20,11 +20,24 @@ from datetime import datetime, timezone
from typing import Any, Dict, Final, Optional
from unittest.mock import AsyncMock, Mock, patch
import fakeredis
import pytest
from fastapi import Request
from pydantic import ValidationError
from litellm.caching.redis_cache import RedisCache
from litellm.proxy.response_polling.polling_handler import ResponsePollingHandler
from tests._support.fake_redis import fake_redis_cache
async def _seed(cache: RedisCache, state: dict[str, object]) -> None:
await cache.async_set_cache(
key=ResponsePollingHandler.get_cache_key(str(state["id"])), value=json.dumps(state), ttl=3600
)
async def _legacy_read(cache: RedisCache, polling_id: str) -> dict[str, object]:
return json.loads(await cache.async_get_cache(ResponsePollingHandler.get_cache_key(polling_id)))
from tests.unit.proxy.router_entry_point_stubs import route_entry_points_to_stubs
@ -58,15 +71,9 @@ class TestResponsePollingHandler:
# OpenAI format
assert ResponsePollingHandler.is_polling_id("resp_abc123") is False
# Anthropic format
assert (
ResponsePollingHandler.is_polling_id("msg_01XFDUDYJgAACzvnptvVoYEL")
is False
)
assert ResponsePollingHandler.is_polling_id("msg_01XFDUDYJgAACzvnptvVoYEL") is False
# Generic UUID
assert (
ResponsePollingHandler.is_polling_id("550e8400-e29b-41d4-a716-446655440000")
is False
)
assert ResponsePollingHandler.is_polling_id("550e8400-e29b-41d4-a716-446655440000") is False
def test_get_cache_key_format(self):
"""Test that cache keys have the correct format"""
@ -80,8 +87,7 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_create_initial_state_returns_queued_status(self):
"""Test that create_initial_state returns response with queued status"""
mock_redis = AsyncMock()
handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600)
handler = ResponsePollingHandler(redis_cache=fake_redis_cache(fakeredis.FakeServer()), ttl=3600)
polling_id = "litellm_poll_test123"
request_data = {
@ -105,8 +111,9 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_create_initial_state_stores_in_redis(self):
"""Test that create_initial_state stores state in Redis with correct TTL"""
mock_redis = AsyncMock()
handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=7200)
server = fakeredis.FakeServer()
cache = fake_redis_cache(server, namespace="ns")
handler = ResponsePollingHandler(redis_cache=cache, ttl=7200)
polling_id = "litellm_poll_test123"
request_data = {"model": "gpt-4o", "input": "Hello"}
@ -116,26 +123,18 @@ class TestResponsePollingHandler:
request_data=request_data,
)
# Verify Redis was called with correct parameters
mock_redis.async_set_cache.assert_called_once()
call_args = mock_redis.async_set_cache.call_args
raw = fakeredis.FakeAsyncRedis(server=server)
assert await raw.keys("*") == [b"ns:litellm:polling:response:litellm_poll_test123"]
assert 7_190_000 < await raw.pttl("ns:litellm:polling:response:litellm_poll_test123") <= 7_200_000
assert (
call_args.kwargs["key"] == "litellm:polling:response:litellm_poll_test123"
)
assert call_args.kwargs["ttl"] == 7200
# Verify the stored value is valid JSON
stored_value = call_args.kwargs["value"]
parsed = json.loads(stored_value)
parsed = await _legacy_read(cache, polling_id)
assert parsed["id"] == polling_id
assert parsed["status"] == "queued"
@pytest.mark.asyncio
async def test_create_initial_state_sets_created_at_timestamp(self):
"""Test that create_initial_state sets a valid created_at timestamp"""
mock_redis = AsyncMock()
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=fake_redis_cache(fakeredis.FakeServer()))
before_time = int(datetime.now(timezone.utc).timestamp())
@ -153,46 +152,45 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_changes_status_to_in_progress(self):
"""Test that update_state can change status to in_progress"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "queued",
"output": [],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600)
handler = ResponsePollingHandler(redis_cache=cache, ttl=3600)
await handler.update_state(
polling_id="litellm_poll_test",
status="in_progress",
)
# Verify the update was saved
mock_redis.async_set_cache.assert_called_once()
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
assert stored["status"] == "in_progress"
@pytest.mark.asyncio
async def test_update_state_replaces_full_output_list(self):
"""Test that update_state replaces the full output list"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
"output": [{"id": "old_item", "type": "message"}],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=3600)
handler = ResponsePollingHandler(redis_cache=cache, ttl=3600)
new_output = [
{
@ -212,8 +210,7 @@ class TestResponsePollingHandler:
output=new_output,
)
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
assert len(stored["output"]) == 2
assert stored["output"][0]["id"] == "item_1"
@ -222,18 +219,19 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_with_usage(self):
"""Test that update_state correctly stores usage data"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
"output": [],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
usage_data = {"input_tokens": 10, "output_tokens": 50, "total_tokens": 60}
@ -243,8 +241,7 @@ class TestResponsePollingHandler:
usage=usage_data,
)
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
assert stored["status"] == "completed"
assert stored["usage"] == usage_data
@ -252,24 +249,23 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_with_reasoning_tools_tool_choice(self):
"""Test that update_state stores reasoning, tools, and tool_choice from response.completed"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
"output": [],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
reasoning_data = {"effort": "medium", "summary": "Step by step analysis"}
tool_choice_data = {"type": "function", "function": {"name": "get_weather"}}
tools_data = [
{"type": "function", "function": {"name": "get_weather", "parameters": {}}}
]
tools_data = [{"type": "function", "function": {"name": "get_weather", "parameters": {}}}]
await handler.update_state(
polling_id="litellm_poll_test",
@ -279,8 +275,7 @@ class TestResponsePollingHandler:
tools=tools_data,
)
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
assert stored["reasoning"] == reasoning_data
assert stored["tool_choice"] == tool_choice_data
@ -289,18 +284,19 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_with_all_responses_api_fields(self):
"""Test that update_state stores all ResponsesAPIResponse fields from response.completed"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
"output": [],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
# All ResponsesAPIResponse fields that can be updated
await handler.update_state(
@ -324,8 +320,7 @@ class TestResponsePollingHandler:
incomplete_details={"reason": "max_output_tokens"},
)
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
# Verify all fields are stored correctly
assert stored["status"] == "completed"
@ -353,8 +348,9 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_preserves_existing_fields(self):
"""Test that update_state preserves fields not being updated"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
@ -363,10 +359,10 @@ class TestResponsePollingHandler:
"created_at": 1234567890,
"model": "gpt-4o",
"temperature": 0.5,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
# Only update status
await handler.update_state(
@ -374,8 +370,7 @@ class TestResponsePollingHandler:
status="completed",
)
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
# Verify existing fields are preserved
assert stored["status"] == "completed"
@ -386,18 +381,19 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_with_error_sets_failed_status(self):
"""Test that providing an error automatically sets status to failed"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
"output": [],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
error_data = {
"type": "internal_error",
@ -410,8 +406,7 @@ class TestResponsePollingHandler:
error=error_data,
)
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
assert stored["status"] == "failed"
assert stored["error"] == error_data
@ -419,18 +414,19 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_with_incomplete_details(self):
"""Test that update_state stores incomplete_details"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
"output": [],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
incomplete_details = {"reason": "max_output_tokens"}
@ -440,8 +436,7 @@ class TestResponsePollingHandler:
incomplete_details=incomplete_details,
)
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
assert stored["status"] == "incomplete"
assert stored["incomplete_details"] == incomplete_details
@ -460,27 +455,23 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_handles_missing_cached_state(self):
"""Test that update_state handles case when cached state doesn't exist"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = None # Cache miss
server = fakeredis.FakeServer()
handler = ResponsePollingHandler(redis_cache=fake_redis_cache(server))
handler = ResponsePollingHandler(redis_cache=mock_redis)
# Should not raise an exception
await handler.update_state(
polling_id="litellm_poll_test",
status="in_progress",
)
# Should not try to set cache if nothing was found
mock_redis.async_set_cache.assert_not_called()
assert await fakeredis.FakeAsyncRedis(server=server).keys("*") == []
# ==================== Get State Tests ====================
@pytest.mark.asyncio
async def test_get_state_returns_cached_state(self):
"""Test that get_state returns the cached state"""
mock_redis = AsyncMock()
cached_state = {
cache = fake_redis_cache(fakeredis.FakeServer())
cached_state: dict[str, object] = {
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
@ -488,9 +479,9 @@ class TestResponsePollingHandler:
"created_at": 1234567890,
"usage": {"input_tokens": 10, "output_tokens": 20},
}
mock_redis.async_get_cache.return_value = json.dumps(cached_state)
await _seed(cache, cached_state)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
result = await handler.get_state("litellm_poll_test")
@ -499,10 +490,7 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_get_state_returns_none_for_missing_state(self):
"""Test that get_state returns None when state doesn't exist"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = None
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=fake_redis_cache(fakeredis.FakeServer()))
result = await handler.get_state("litellm_poll_nonexistent")
@ -522,25 +510,25 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_cancel_polling_updates_status_to_cancelled(self):
"""Test that cancel_polling sets status to cancelled"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
"output": [],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
result = await handler.cancel_polling("litellm_poll_test")
assert result is True
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
assert stored["status"] == "cancelled"
# ==================== Delete Polling Tests ====================
@ -548,27 +536,38 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_delete_polling_removes_from_cache(self):
"""Test that delete_polling removes the entry from Redis"""
mock_redis = AsyncMock()
mock_async_client = AsyncMock()
mock_redis.redis_async_client = True # hasattr check
# init_async_client is a sync method that returns an async client
mock_redis.init_async_client = Mock(return_value=mock_async_client)
# Mock async_delete_cache to actually call init_async_client and delete
async def mock_async_delete_cache(key):
client = mock_redis.init_async_client()
await client.delete(key)
mock_redis.async_delete_cache = mock_async_delete_cache
handler = ResponsePollingHandler(redis_cache=mock_redis)
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(cache, {"id": "litellm_poll_test", "status": "queued"})
handler = ResponsePollingHandler(redis_cache=cache)
result = await handler.delete_polling("litellm_poll_test")
assert result is True
mock_async_client.delete.assert_called_once_with(
"litellm:polling:response:litellm_poll_test"
assert await handler.get_state("litellm_poll_test") is None
@pytest.mark.asyncio
@pytest.mark.parametrize("stored", [b"not json", b'{"id": "single-encoded"}', b'"[1, 2]"'])
async def test_a_state_that_does_not_decode_reads_as_missing(self, stored: bytes):
server = fakeredis.FakeServer()
await fakeredis.FakeAsyncRedis(server=server).set(
ResponsePollingHandler.get_cache_key("litellm_poll_x"), stored
)
handler = ResponsePollingHandler(redis_cache=fake_redis_cache(server))
assert await handler.get_state("litellm_poll_x") is None
@pytest.mark.asyncio
async def test_an_unreachable_redis_reads_as_missing_and_fails_the_delete(self):
server = fakeredis.FakeServer()
cache = fake_redis_cache(server)
await _seed(cache, {"id": "litellm_poll_test", "status": "queued"})
handler = ResponsePollingHandler(redis_cache=cache)
server.connected = False
assert await handler.get_state("litellm_poll_test") is None
assert await handler.delete_polling("litellm_poll_test") is False
queued = await handler.create_initial_state(polling_id="litellm_poll_new", request_data={})
assert queued.status == "queued"
@pytest.mark.asyncio
async def test_delete_polling_returns_false_without_redis(self):
@ -596,26 +595,30 @@ class TestResponsePollingHandler:
@pytest.mark.asyncio
async def test_update_state_uses_configured_ttl(self):
"""Test that update_state uses the configured TTL"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
server = fakeredis.FakeServer()
cache = fake_redis_cache(server)
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "queued",
"output": [],
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis, ttl=1800)
handler = ResponsePollingHandler(redis_cache=cache, ttl=1800)
await handler.update_state(
polling_id="litellm_poll_test",
status="in_progress",
)
call_args = mock_redis.async_set_cache.call_args
assert call_args.kwargs["ttl"] == 1800
ttl_ms = await fakeredis.FakeAsyncRedis(server=server).pttl(
ResponsePollingHandler.get_cache_key("litellm_poll_test")
)
assert 1_790_000 < ttl_ms <= 1_800_000
class TestStreamingEventProcessing:
@ -689,6 +692,7 @@ class TestBackgroundStreamingModule:
def test_background_streaming_task_is_async(self):
"""Test that background_streaming_task is an async function"""
import inspect
from litellm.proxy.response_polling.background_streaming import (
background_streaming_task,
)
@ -886,9 +890,7 @@ class TestProviderResolutionForPolling:
model_name_to_deployment_indices = {
"gpt-5": [0],
}
model_list = [
{"model_name": "gpt-5", "litellm_params": {"model": "openai/gpt-5"}}
]
model_list = [{"model_name": "gpt-5", "litellm_params": {"model": "openai/gpt-5"}}]
model = "unknown-model" # Not in router
polling_via_cache_enabled = ["openai"]
@ -1282,9 +1284,7 @@ class TestStreamingEventParsing:
content_list = output_items[item_id]["content"]
if content_index < len(content_list):
if isinstance(content_list[content_index], dict):
content_list[content_index]["text"] = accumulated_text[
key
]
content_list[content_index]["text"] = accumulated_text[key]
assert accumulated_text[("item_123", 0)] == "Hello World!"
assert output_items["item_123"]["content"][0]["text"] == "Hello World!"
@ -1346,9 +1346,7 @@ class TestStreamingEventParsing:
def test_parse_sse_format(self):
"""Test parsing Server-Sent Events format"""
raw_chunk = (
b'data: {"type": "response.output_item.added", "item": {"id": "123"}}'
)
raw_chunk = b'data: {"type": "response.output_item.added", "item": {"id": "123"}}'
# Decode bytes to string
if isinstance(raw_chunk, bytes):
@ -1476,9 +1474,7 @@ class TestBackgroundStreamingTerminalEvents:
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
MockProcessor.return_value.base_process_llm_request = AsyncMock(return_value=mock_response)
await background_streaming_task(**kwargs)
# Find the final update_state call (last one)
@ -1515,9 +1511,7 @@ class TestBackgroundStreamingTerminalEvents:
with patch( # test-quality-ok: the processor is built inside the task, same idiom as the sibling tests
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
MockProcessor.return_value.base_process_llm_request = AsyncMock(return_value=mock_response)
await background_streaming_task(**kwargs)
final_call = handler.update_state.call_args_list[-1]
@ -1558,17 +1552,13 @@ class TestBackgroundStreamingTerminalEvents:
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
MockProcessor.return_value.base_process_llm_request = AsyncMock(return_value=mock_response)
await background_streaming_task(**kwargs)
final_call = handler.update_state.call_args_list[-1]
assert final_call.kwargs["status"] == "incomplete"
assert final_call.kwargs["error"] == error_payload
assert final_call.kwargs["incomplete_details"] == {
"reason": "max_output_tokens"
}
assert final_call.kwargs["incomplete_details"] == {"reason": "max_output_tokens"}
assert final_call.kwargs["usage"] == {"input_tokens": 10, "output_tokens": 4096}
@pytest.mark.asyncio
@ -1597,9 +1587,7 @@ class TestBackgroundStreamingTerminalEvents:
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
MockProcessor.return_value.base_process_llm_request = AsyncMock(return_value=mock_response)
await background_streaming_task(**kwargs)
final_call = handler.update_state.call_args_list[-1]
@ -1632,9 +1620,7 @@ class TestBackgroundStreamingTerminalEvents:
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
MockProcessor.return_value.base_process_llm_request = AsyncMock(return_value=mock_response)
await background_streaming_task(**kwargs)
final_call = handler.update_state.call_args_list[-1]
@ -1699,9 +1685,7 @@ class TestBackgroundStreamingTerminalEvents:
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
MockProcessor.return_value.base_process_llm_request = AsyncMock(return_value=mock_response)
await background_streaming_task(**kwargs)
final_call = handler.update_state.call_args_list[-1]
@ -1725,9 +1709,7 @@ class TestBackgroundStreamingTerminalEvents:
with patch(
"litellm.proxy.response_polling.background_streaming.ProxyBaseLLMRequestProcessing"
) as MockProcessor:
MockProcessor.return_value.base_process_llm_request = AsyncMock(
return_value=mock_response
)
MockProcessor.return_value.base_process_llm_request = AsyncMock(return_value=mock_response)
await background_streaming_task(**kwargs)
final_call = handler.update_state.call_args_list[-1]
@ -1763,9 +1745,7 @@ class TestBackgroundStreamingTerminalEvents:
yield f"data: {json.dumps(event)}\n\n"
async def upstream_call_behind_disconnect_guard(**kwargs):
return await create_response(
slow_upstream_stream(), "text/event-stream", {}, request=kwargs["request"]
)
return await create_response(slow_upstream_stream(), "text/event-stream", {}, request=kwargs["request"])
handler = AsyncMock(spec=ResponsePollingHandler)
kwargs = _make_background_streaming_kwargs("poll_7", handler)
@ -1838,8 +1818,7 @@ class TestEdgeCases:
@pytest.mark.asyncio
async def test_create_initial_state_with_empty_metadata(self):
"""Test create_initial_state handles missing metadata gracefully"""
mock_redis = AsyncMock()
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=fake_redis_cache(fakeredis.FakeServer()))
response = await handler.create_initial_state(
polling_id="litellm_poll_test",
@ -1851,26 +1830,26 @@ class TestEdgeCases:
@pytest.mark.asyncio
async def test_update_state_with_none_output_clears_output(self):
"""Test that output=[] explicitly sets empty output"""
mock_redis = AsyncMock()
mock_redis.async_get_cache.return_value = json.dumps(
cache = fake_redis_cache(fakeredis.FakeServer())
await _seed(
cache,
{
"id": "litellm_poll_test",
"object": "response",
"status": "in_progress",
"output": [{"id": "item_1"}], # Has existing output
"created_at": 1234567890,
}
},
)
handler = ResponsePollingHandler(redis_cache=mock_redis)
handler = ResponsePollingHandler(redis_cache=cache)
await handler.update_state(
polling_id="litellm_poll_test",
output=[], # Explicitly set empty
)
call_args = mock_redis.async_set_cache.call_args
stored = json.loads(call_args.kwargs["value"])
stored = await _legacy_read(cache, "litellm_poll_test")
assert stored["output"] == []

174
tests/unit/redis/transport/test_redis_py.py
View file

@ -4,12 +4,17 @@ from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Final, Literal
from unittest.mock import patch
import fakeredis
import pytest
import litellm
from litellm._internal_context import current_service_target, service_target
from litellm._service_logger import ServiceLogging
from litellm.redis.transport.contract import Delete, Get, RedisFailure, Set
from litellm.caching.redis_cluster_cache import ClusterConnectionGate, RedisClusterCache
from litellm.redis.transport.contract import RedisFailure
from litellm.types.services import ServiceTypes
from tests._support.fake_redis import fake_redis_cache, fake_redis_transport
@ -20,12 +25,12 @@ async def test_every_command_prefixes_the_namespace_once(
) -> None:
transport = fake_redis_cache(server, namespace="ns").transport()
assert await transport.execute(Set("k", "v", ttl=None)) is True
assert await transport.execute(Set("ns:gone", "v", ttl=None)) is True
assert await transport.set("k", "v", ttl=None) is True
assert await transport.set("ns:gone", "v", ttl=None) is True
assert sorted(await raw.keys("*")) == [b"ns:gone", b"ns:k"]
assert await transport.execute(Get("k")) == await transport.execute(Get("ns:k")) == b"v"
assert await transport.execute(Delete("gone")) == 1
assert await transport.get("k") == await transport.get("ns:k") == b"v"
assert await transport.delete("gone") == 1
assert await raw.keys("*") == [b"ns:k"]
@ -35,8 +40,8 @@ async def test_set_nx_refuses_a_live_key_and_applies_the_ttl(
) -> None:
transport = fake_redis_cache(server).transport()
assert await transport.execute(Set("k", "first", ttl=timedelta(seconds=30), nx=True)) is True
assert await transport.execute(Set("k", "second", ttl=timedelta(seconds=30), nx=True)) is False
assert await transport.set("k", "first", ttl=timedelta(seconds=30), nx=True) is True
assert await transport.set("k", "second", ttl=timedelta(seconds=30), nx=True) is False
assert await raw.get("k") == b"first"
assert 0 < await raw.pttl("k") <= 30_000
@ -46,7 +51,7 @@ async def test_set_nx_refuses_a_live_key_and_applies_the_ttl(
async def test_set_without_ttl_persists(server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis) -> None:
transport = fake_redis_cache(server).transport()
assert await transport.execute(Set("k", "v", ttl=None)) is True
assert await transport.set("k", "v", ttl=None) is True
assert await raw.pttl("k") == -1
@ -58,8 +63,8 @@ async def test_delete_reports_how_many_keys_it_removed(
transport = fake_redis_cache(server).transport()
await raw.set("k", "v")
assert await transport.execute(Delete("k")) == 1
assert await transport.execute(Delete("k")) == 0
assert await transport.delete("k") == 1
assert await transport.delete("k") == 0
assert await raw.exists("k") == 0
@ -70,11 +75,11 @@ async def test_a_rejected_command_is_a_server_failure_and_keeps_the_connection(
transport = fake_redis_cache(server).transport()
await raw.rpush("list", "x")
rejected = await transport.execute(Get("list"))
rejected = await transport.get("list")
assert isinstance(rejected, RedisFailure)
assert rejected.kind == "server"
assert await transport.execute(Set("k", "v", ttl=None)) is True
assert await transport.set("k", "v", ttl=None) is True
@pytest.mark.asyncio
@ -82,7 +87,7 @@ async def test_an_unreachable_server_is_a_transport_failure(server: fakeredis.Fa
transport = fake_redis_cache(server).transport()
server.connected = False
failure = await transport.execute(Get("k"))
failure = await transport.get("k")
assert isinstance(failure, RedisFailure)
assert failure.kind == "transport"
@ -95,9 +100,9 @@ async def test_a_transport_failure_opens_the_breaker_and_it_answers_without_redi
transport = fake_redis_transport(raw, failure_threshold=1)
server.connected = False
tripped = await transport.execute(Get("k"))
tripped = await transport.get("k")
server.connected = True
refused = await transport.execute(Get("k"))
refused = await transport.get("k")
assert isinstance(tripped, RedisFailure)
assert tripped.kind == "transport"
@ -105,14 +110,112 @@ async def test_a_transport_failure_opens_the_breaker_and_it_answers_without_redi
assert refused.kind == "breaker_open"
@pytest.mark.asyncio
async def test_mget_returns_one_slot_per_key_in_order(
server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
) -> None:
transport = fake_redis_cache(server, namespace="ns").transport()
await raw.set("ns:a", "1")
await raw.set("ns:c", "3")
assert await transport.mget(("a", "b", "ns:c")) == (b"1", None, b"3")
assert await transport.mget(()) == ()
@pytest.mark.asyncio
async def test_delete_takes_several_keys(server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis) -> None:
transport = fake_redis_cache(server, namespace="ns").transport()
await raw.set("ns:a", "1")
await raw.set("ns:b", "2")
await raw.set("c", "3")
assert await transport.delete("a", "b", "c") == 2
assert await raw.keys("*") == [b"c"]
_ECHO: Final = """
redis.call("set", KEYS[1], ARGV[1])
if ARGV[2] == "false" then
return false
end
return {redis.call("get", KEYS[1]), tonumber(ARGV[2]), {KEYS[1]}}
"""
@pytest.mark.asyncio
async def test_eval_namespaces_keys_and_returns_nested_replies(
server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis
) -> None:
transport = fake_redis_cache(server, namespace="ns").transport()
assert await transport.eval(_ECHO, ("k",), ("v", 7)) == (b"v", 7, (b"ns:k",))
assert await transport.eval(_ECHO, ("k",), ("w", "false")) is None
assert await raw.get("ns:k") == b"w"
@pytest.mark.asyncio
async def test_eval_survives_a_script_flush(server: fakeredis.FakeServer, raw: fakeredis.FakeAsyncRedis) -> None:
transport = fake_redis_cache(server).transport()
assert await transport.eval(_ECHO, ("k",), ("v", 1)) == (b"v", 1, (b"k",))
await raw.script_flush()
assert await transport.eval(_ECHO, ("k",), ("w", 2)) == (b"w", 2, (b"k",))
@pytest.mark.asyncio
async def test_a_script_error_is_a_server_failure(server: fakeredis.FakeServer) -> None:
transport = fake_redis_cache(server).transport()
failure = await transport.eval('return redis.error_reply("nope")', (), ())
assert isinstance(failure, RedisFailure)
assert failure.kind == "server"
assert await transport.eval("return 1", (), ()) == 1
class _OneAtATimeClient:
def __init__(self, inner: fakeredis.FakeAsyncRedis) -> None:
self.inner = inner
self.in_flight = 0
self.peak_in_flight = 0
async def get(self, name: str) -> object:
self.in_flight += 1
self.peak_in_flight = max(self.peak_in_flight, self.in_flight)
await asyncio.sleep(0.01)
self.in_flight -= 1
return await self.inner.get(name)
@pytest.mark.asyncio
async def test_a_cluster_cache_transport_waits_on_the_connection_gate(raw: fakeredis.FakeAsyncRedis) -> None:
with patch("litellm._redis.init_redis_cluster"):
cache = RedisClusterCache(
startup_nodes=[{"host": "localhost", "port": 7000}],
connection_gate=ClusterConnectionGate(limit=1, wait_timeout=5.0),
)
client = _OneAtATimeClient(raw)
litellm.in_memory_llm_clients_cache.set_cache(
key=cache._get_async_client_cache_key(), # pyright: ignore[reportPrivateUsage] # seeds the per-loop client slot RedisCache reads
value=client,
)
await raw.set("k", "v")
transport = cache.transport()
replies = await asyncio.gather(*(transport.get("k") for _ in range(5)))
assert replies == [b"v"] * 5
assert client.peak_in_flight == 1
@pytest.mark.asyncio
async def test_replies_stay_bytes_on_a_client_that_decodes_responses(server: fakeredis.FakeServer) -> None:
transport = fake_redis_transport(
fakeredis.FakeAsyncRedis(server=server, decode_responses=True), failure_threshold=1
)
assert await transport.execute(Set("k", "v", ttl=None)) is True
assert await transport.execute(Get("k")) == b"v"
assert await transport.set("k", "v", ttl=None) is True
assert await transport.get("k") == b"v"
async def _settle() -> None:
@ -135,6 +238,9 @@ class _RecordingEvents(ServiceLogging):
super().__init__()
self.seen: tuple[_Event, ...] = ()
def has_sinks(self) -> bool:
return True
async def async_service_success_hook(
self,
service: ServiceTypes,
@ -173,11 +279,10 @@ async def test_every_command_reports_one_service_event_under_the_callers_target(
transport = fake_redis_transport(raw, failure_threshold=2, events=events)
with service_target("response_polling"):
await transport.execute(Set("k", "v", ttl=None))
await transport.execute(Get("k"))
await transport.set("k", "v", ttl=None)
await transport.get("k")
server.connected = False
failed = await transport.execute(Delete("k"))
await asyncio.sleep(0)
failed = await transport.delete("k")
assert isinstance(failed, RedisFailure)
assert events.seen == (
@ -195,9 +300,8 @@ async def test_a_call_the_breaker_refuses_reports_no_service_event(
transport = fake_redis_transport(raw, failure_threshold=1, events=events)
server.connected = False
await transport.execute(Get("k"))
refused = await transport.execute(Get("k"))
await asyncio.sleep(0)
await transport.get("k")
refused = await transport.get("k")
assert isinstance(refused, RedisFailure) and refused.kind == "breaker_open"
assert tuple(event.outcome for event in events.seen) == ("failure",)
@ -205,18 +309,32 @@ async def test_a_call_the_breaker_refuses_reports_no_service_event(
@pytest.mark.asyncio
async def test_the_cache_transport_reports_to_the_caches_service_logger(server: fakeredis.FakeServer) -> None:
events = _RecordingEvents()
transport = fake_redis_cache(server, service_logger=events).transport()
await _settle()
before = events.seen
await transport.get("k")
server.connected = False
await transport.get("k")
assert tuple(event.outcome for event in events.seen[len(before) :]) == ("success", "failure")
@pytest.mark.asyncio
async def test_without_a_service_callback_no_event_is_built(server: fakeredis.FakeServer) -> None:
service_logger = ServiceLogging(mock_testing=True)
transport = fake_redis_cache(server, service_logger=service_logger).transport()
await _settle()
before = (service_logger.mock_testing_async_success_hook, service_logger.mock_testing_async_failure_hook)
await transport.execute(Get("k"))
assert await transport.get("k") is None
server.connected = False
await transport.execute(Get("k"))
assert isinstance(await transport.get("k"), RedisFailure)
await _settle()
after = (service_logger.mock_testing_async_success_hook, service_logger.mock_testing_async_failure_hook)
assert (after[0] - before[0], after[1] - before[1]) == (1, 1)
assert after == before
@pytest.mark.asyncio
@ -229,7 +347,7 @@ async def test_a_failure_is_logged_once_with_the_command_and_target(
caplog.clear()
with caplog.at_level(logging.WARNING), service_target("response_polling"):
failure = await transport.execute(Get("k"))
failure = await transport.get("k")
assert isinstance(failure, RedisFailure)
assert [record.getMessage() for record in caplog.records] == [

72
uv.lock generated
View file

@ -2123,6 +2123,11 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/49/b5/82f89307d0d769cd9bf46a54fb9136be08e4e57c5570ae421db4c9a2ba62/fakeredis-2.34.1-py3-none-any.whl", hash = "sha256:0107ec99d48913e7eec2a5e3e2403d1bd5f8aa6489d1a634571b975289c48f12", size = 122160, upload-time = "2026-02-25T13:17:49.701Z" },
]
[package.optional-dependencies]
lua = [
{ name = "lupa" },
]
[[package]]
name = "fastapi"
version = "0.136.3"
@ -5148,7 +5153,7 @@ dev = [
{ name = "basedpyright" },
{ name = "botocore-stubs" },
{ name = "diff-cover" },
{ name = "fakeredis" },
{ name = "fakeredis", extra = ["lua"] },
{ name = "fastapi-offline" },
{ name = "hypothesis" },
{ name = "keyring" },
@ -5377,7 +5382,7 @@ dev = [
{ name = "basedpyright", specifier = "==1.39.7" },
{ name = "botocore-stubs", specifier = "==1.43.14" },
{ name = "diff-cover", specifier = "==9.7.2" },
{ name = "fakeredis", specifier = "==2.34.1" },
{ name = "fakeredis", extras = ["lua"], specifier = "==2.34.1" },
{ name = "fastapi-offline", specifier = "==1.7.6" },
{ name = "hypothesis", specifier = "==6.165.10" },
{ name = "keyring", specifier = "==25.7.0" },
@ -5602,6 +5607,69 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/eb/cf/e4e016820516c1f0c85549e356865c2319cf6afba8ab386611b1d03cf2b6/lunary-1.4.37-py3-none-any.whl", hash = "sha256:7289330e851a481404c92213ce480c0c7be9bfa56982a1385e08e3665abefe05", size = 25581, upload-time = "2026-02-12T08:15:03.892Z" },
]
[[package]]
name = "lupa"
version = "2.8"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/c3/a6/0f869fbb07c393f15473b1eefefb7b5bec162fb7481803d040ed4dc46002/lupa-2.8.tar.gz", hash = "sha256:d8022641b9ec8ecf2c5ecbe9f47e5a70e0b87c4b5ae921b92cb02a638e0acd08", size = 6156370, upload-time = "2026-04-15T20:08:30.534Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/09/21/9be4516ddd22f8eadba336d9ba065d17d79108465ae1b7f71424ab99b9d0/lupa-2.8-cp310-abi3-win32.whl", hash = "sha256:c2a5fd15dc62374e1661a55f01744c9ec1c56f291ba4a0749d3af2174556e78f", size = 1594887, upload-time = "2026-04-15T20:05:23.377Z" },
{ url = "https://files.pythonhosted.org/packages/2d/99/1557c9685d7034d9ce8dd2b54c40a26d6deb7c67c1fdb5c801abd1a02c3f/lupa-2.8-cp310-abi3-win_arm64.whl", hash = "sha256:9e304fb1c50cf23fd8882afbe1aa87525ef8a72667bcab3b37b2bbb2bc542269", size = 1371742, upload-time = "2026-04-15T20:05:27.417Z" },
{ url = "https://files.pythonhosted.org/packages/1c/34/05ce4745b191633f90ff1ab50f1a19a37da282bb0a41fb500d9157fc9b8f/lupa-2.8-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:97bd01e90b8031e56a5fd5bb70605aea09f1dba675c1140308a52780f93d06f1", size = 1202714, upload-time = "2026-04-15T20:05:31.088Z" },
{ url = "https://files.pythonhosted.org/packages/7d/d2/f70fdbeec2d4c69ee6a469e6cddde9635fff4af4e13fb652e6a1229eef51/lupa-2.8-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0b5ebe1a13c45767919c86750b84fe2da9f6288b6f3cea4ce7660bb2abc9d921", size = 1857453, upload-time = "2026-04-15T20:05:34.611Z" },
{ url = "https://files.pythonhosted.org/packages/97/dc/6fcda0e36e75eb6cb98dc9190fa4737d727eeae29e58f892980b2c96b656/lupa-2.8-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:097e7d0f1719a88020b67c82e05d53d7973c166952393afcecfd8434c7e19a15", size = 2408890, upload-time = "2026-04-15T20:05:37.994Z" },
{ url = "https://files.pythonhosted.org/packages/58/29/7ea176eac3c1dac83d059762daa875ad1390decc0bf2c3b4c7bbfc1f1665/lupa-2.8-cp310-cp310-win_amd64.whl", hash = "sha256:7bb223ee8f72d0dc076b0d65296ee72f1c69450f9d2fed5315f7707d98c4a03d", size = 1910396, upload-time = "2026-04-15T20:05:41.163Z" },
{ url = "https://files.pythonhosted.org/packages/b7/0a/5a740717f27aa77481e6a61b97cf79d1e0c1ede729b1268caacded915326/lupa-2.8-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:b12e43c1fb787189dfc28cd604aef0baa2cb95e27da19498d520361d0ace070a", size = 1202376, upload-time = "2026-04-15T20:05:44.049Z" },
{ url = "https://files.pythonhosted.org/packages/1b/75/6b64d0098c64275a801896cb7a6a30e7e653d25fa102c64e747292afcdbb/lupa-2.8-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:f6f603391dffb256e36a79fd2044084d5f4b8a0a4c0e5ad291cd3ab3aaf1fd0a", size = 1839271, upload-time = "2026-04-15T20:05:47.399Z" },
{ url = "https://files.pythonhosted.org/packages/7b/2f/0d4f00563046ff616ef6a421f8b776a5ffb327f7b32ed69e856d52b917a8/lupa-2.8-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9f6f41c91366e7d0d474f87d81c1274af861f40812bf729c9f97ab4c8f3c7ac8", size = 2376251, upload-time = "2026-04-15T20:05:49.891Z" },
{ url = "https://files.pythonhosted.org/packages/4c/8e/caa83237f427d9e85b7f02c816e7270c9c9571dec1673e06b0180402f70e/lupa-2.8-cp311-cp311-win_amd64.whl", hash = "sha256:f5a6af145b0ea818f01d27bfe2583a4b538570bef61d22c8773e0eccf011234c", size = 1923488, upload-time = "2026-04-15T20:05:52.954Z" },
{ url = "https://files.pythonhosted.org/packages/ad/0b/368f2f0bc750b25c69d4563e44f677925ab5dd3d2887f9b0c15465d21a2a/lupa-2.8-cp312-abi3-macosx_10_13_x86_64.whl", hash = "sha256:f4342f4de76ae7ce2ab0672d36003bdb7e1a33252f293b569298ddd792e70e33", size = 1194056, upload-time = "2026-04-15T20:05:55.794Z" },
{ url = "https://files.pythonhosted.org/packages/5b/0f/c89eb8dd36fdea4e50ae3f7f5275bea3b0cc5d4057b8ee7b3bbc78010422/lupa-2.8-cp312-abi3-manylinux2010_i686.manylinux_2_12_i686.manylinux_2_28_i686.whl", hash = "sha256:4203fa1659315e939a5304e75001b8cc14234fb3cbb3ed86c049b0cc5d90fcee", size = 1434278, upload-time = "2026-04-15T20:05:57.94Z" },
{ url = "https://files.pythonhosted.org/packages/47/30/c3b4d2cd8733621b404b8a4214e5f852955c4ba632546dc84123bea9ee89/lupa-2.8-cp312-abi3-manylinux2014_armv7l.manylinux_2_17_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:81f2d843ce668b653146c007467570210ae44be51dac6926666c51d49536f307", size = 1150068, upload-time = "2026-04-15T20:06:01.04Z" },
{ url = "https://files.pythonhosted.org/packages/8d/d2/bac12c398519efafc6af84be1974edd0d7a4895fb4735b5c8d615d298595/lupa-2.8-cp312-abi3-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:d3d0cde2c77588d1c60875a4f34f059513476c6e1775351897195b51e0f3df08", size = 1409532, upload-time = "2026-04-15T20:06:03.592Z" },
{ url = "https://files.pythonhosted.org/packages/9c/6a/18b52e11962014026e07813530b0b108ee8bc0a2a13ef0eaea5d41dce023/lupa-2.8-cp312-abi3-manylinux_2_34_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:9e0d11b8f3a8dac6413f704fef7161d048bb10c58bdac6cbffa5e60efa56e9a3", size = 1242687, upload-time = "2026-04-15T20:06:06.863Z" },
{ url = "https://files.pythonhosted.org/packages/b3/8e/7fd4eb049875f61429b96780d2eae4700f0e78fe0a52db8edb231b1cd09f/lupa-2.8-cp312-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:54cff414f21f8cd8c6be4aae52541f3b9cd39602b59e3a3db9b5c9f9f674ff18", size = 1856038, upload-time = "2026-04-15T20:06:09.358Z" },
{ url = "https://files.pythonhosted.org/packages/e9/f9/37ad9d2773d30f2931890d310a4bdce28d45484206e6f48bc18b0325eabd/lupa-2.8-cp312-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:24b4d8af5558e549b70daf1547f5c1c1d664ecea9fc790f83efe5d75e9a93797", size = 1128982, upload-time = "2026-04-15T20:06:12.312Z" },
{ url = "https://files.pythonhosted.org/packages/57/31/c0fd7984c24844ea79caa45c0235f61a06b38fd69a839f6c62770f8d684a/lupa-2.8-cp312-abi3-musllinux_1_2_i686.whl", hash = "sha256:ce86dff1ee7f7cf45f5622065ae991949dd7bb1703581cbc58a630137bb7ccf9", size = 1457594, upload-time = "2026-04-15T20:06:15.881Z" },
{ url = "https://files.pythonhosted.org/packages/11/f5/a28e411be30ec1bf0db1eb0c087eebc73be9e7a1adcfe6ac209861ccc446/lupa-2.8-cp312-abi3-musllinux_1_2_ppc64le.whl", hash = "sha256:f4d01b2a08c70bbb883a9e082b6b36b89121ed5910b710f1ba11c73295ff4fba", size = 1425721, upload-time = "2026-04-15T20:06:18.009Z" },
{ url = "https://files.pythonhosted.org/packages/ed/c1/359f767c4ae024be30d909fe8a9f0e9af266bad47ce2bd2ed248fb986fcf/lupa-2.8-cp312-abi3-musllinux_1_2_riscv64.whl", hash = "sha256:7f210d5a8353e510ea1199c42cf3cbdd630553bf2bc8fb4c00fea06fdec7c798", size = 1253258, upload-time = "2026-04-15T20:06:21.17Z" },
{ url = "https://files.pythonhosted.org/packages/17/52/473f11790c261fd02bbf318a546fe040e9ec9f677181272fa78d3b4112a4/lupa-2.8-cp312-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:4f81a02806e7c7ad26d8c6fa222c8bef1b0c1b124347c879be880b41339d41e4", size = 2395272, upload-time = "2026-04-15T20:06:24.137Z" },
{ url = "https://files.pythonhosted.org/packages/94/bf/75c8795655a8836eab6a11a630352c4b7c5dc5c54d075077bc9bffdeee45/lupa-2.8-cp312-abi3-win32.whl", hash = "sha256:360056453a7a4eaa4ac5a204c31a5a014b1eb2ee5490603234d2ba831684f1f2", size = 1606136, upload-time = "2026-04-15T20:06:27.815Z" },
{ url = "https://files.pythonhosted.org/packages/d8/29/11a2cdd612b6f55e506292dfb6ba343216e80a693e7fe3f876ef204ce9c6/lupa-2.8-cp312-abi3-win_arm64.whl", hash = "sha256:1628371c6592a6d5650497a9e31fb2bb3a7e9883c1f301d1111265e484045af9", size = 1364495, upload-time = "2026-04-15T20:06:30.254Z" },
{ url = "https://files.pythonhosted.org/packages/4d/17/fa834b6b09ad17e7df5d0f7715d64877a125a3776ada689751a1f9dc2959/lupa-2.8-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:450650f91c48c2415b0d59ab3abfcfda3b6efb5b858205f4d4bda8ad141fa529", size = 1190111, upload-time = "2026-04-15T20:06:32.84Z" },
{ url = "https://files.pythonhosted.org/packages/ab/43/45589901b7d1a0e3a9d91d19a311fb6a56924e8571536c3f2212160fd953/lupa-2.8-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:27044f3363047f946b3d3aab9157cbd172b3538ada9ec1baef43432bf7d03a78", size = 1812999, upload-time = "2026-04-15T20:06:35.664Z" },
{ url = "https://files.pythonhosted.org/packages/a1/ac/4ade7d15ff5c61758d7943ac6f0a496bf1cc65b6c09f842b52a0702e664c/lupa-2.8-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8cf4f064a0e5531afce2d7d750120c10c10f9529139af6ca6150d13151034398", size = 2368731, upload-time = "2026-04-15T20:06:37.959Z" },
{ url = "https://files.pythonhosted.org/packages/0c/27/05f950d15b8ab120b39c43588b438ff3ace70c1b1b0225a960393a497483/lupa-2.8-cp312-cp312-win_amd64.whl", hash = "sha256:281bedc5deb92d31e649a3552edd662449365a635904fa4d5cb4509c7245e34e", size = 1941809, upload-time = "2026-04-15T20:06:40.302Z" },
{ url = "https://files.pythonhosted.org/packages/a6/3f/19f83c3a0c84dc8bea8a58e7416dca6a3ede662c33c8d1ec758e5afc754a/lupa-2.8-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:45fc9da0145ecb0083ef5ff9975116cc784bd0258bdc2bd131ba15483ce18398", size = 1201203, upload-time = "2026-04-15T20:06:42.169Z" },
{ url = "https://files.pythonhosted.org/packages/89/0f/a14f0073f09610158038582e230618a48c14da6bd88185289461aa4cb854/lupa-2.8-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:58e18afed57955b41130e269c78f53d4123ab86e236b53816f4cbffa25cb5d30", size = 1806210, upload-time = "2026-04-15T20:06:45.486Z" },
{ url = "https://files.pythonhosted.org/packages/2f/14/48fff156c63a136001a7620878af7d31aa07e66b495ed621e3eddd73c294/lupa-2.8-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:fc47f536ac13a79cef47d29a2b205576a22841f042a2bcec1676b95806e7706a", size = 2359005, upload-time = "2026-04-15T20:06:47.819Z" },
{ url = "https://files.pythonhosted.org/packages/fe/18/3ac638ec90edf178242b8a2b2f00f8adae694248c03a26341ef941bb746e/lupa-2.8-cp313-cp313-win_amd64.whl", hash = "sha256:ce9404c661dbac65cc9bed351ad45e797af93d30d70be309a3fa8209ac86d93b", size = 1936754, upload-time = "2026-04-15T20:06:50.448Z" },
{ url = "https://files.pythonhosted.org/packages/b0/ef/5ee5fed6ea7459a671196359ce04bfeeaf26be1dac8ff24bf28e5c7a6e81/lupa-2.8-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:348c3f8ecabb6324dcbc05c2740d762ef8fcec7b06c79e45262ab97a217684e3", size = 1209388, upload-time = "2026-04-15T20:06:53.022Z" },
{ url = "https://files.pythonhosted.org/packages/6e/b1/67a940d5542cb0384b443fe951b5a83ea9340d1333a733a258fdd1c619ba/lupa-2.8-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:951496471056061598a7d1729a6cdf48d662fec777a9f2d8aa5a1e62fd30e5a5", size = 1826821, upload-time = "2026-04-15T20:06:55.699Z" },
{ url = "https://files.pythonhosted.org/packages/a1/a2/b354e5ba3b911ec50686003dc8897e892b9e8c5c036b33219b03d54c4daf/lupa-2.8-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:a591b9947ca347b41a63370e121d6e2b1458fe6dde9ae065029ec10a37f25ff4", size = 2366893, upload-time = "2026-04-15T20:06:58.9Z" },
{ url = "https://files.pythonhosted.org/packages/8e/52/d76066401f29539df5352f70ecded66576f32933b6045cd0bfc56cb770b9/lupa-2.8-cp314-cp314-win_amd64.whl", hash = "sha256:3903c9cf628dae2f56405503247b77a61a3a61bd2dda470e336950c74776d55d", size = 1994716, upload-time = "2026-04-15T20:07:19.194Z" },
{ url = "https://files.pythonhosted.org/packages/c3/bd/3efc437a4361c16d25e66478c50357c9a8e8ecfb718fe749eb9ca3176ef6/lupa-2.8-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:f711a8ab0486b9ac6fdda94a22ddcfbc9f0d4a27e3a8cf1bf79c6e48b33017c1", size = 1251217, upload-time = "2026-04-15T20:07:01.64Z" },
{ url = "https://files.pythonhosted.org/packages/ea/f4/2e9f8ecbaca854bfdf14af8a9b505ec0cbc640377b3b218921594b7563cd/lupa-2.8-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:dc51250e76367a3e27fcd01dc769b9bfcbbc34f48df48dde53d6af6e75b7eaa5", size = 1814701, upload-time = "2026-04-15T20:07:04.149Z" },
{ url = "https://files.pythonhosted.org/packages/ba/53/4000b1acaa8b1f3827fcff0cfcdff44d3befddda42cab7e685a49689b5a1/lupa-2.8-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f8a22088a552828958603323f0a5c4b3e11e03b75d0bf4c965ef879de9b60a8d", size = 2348414, upload-time = "2026-04-15T20:07:07.285Z" },
{ url = "https://files.pythonhosted.org/packages/d5/78/26ee48d3890cddf03cefb65f433e3492759c0b3c0582180755bddbaab7bd/lupa-2.8-cp314-cp314t-win32.whl", hash = "sha256:4f7c553c1d8cfffbe85d81daef730d12cae4b6002d457542914da0ac8a1145b3", size = 1831611, upload-time = "2026-04-15T20:07:09.752Z" },
{ url = "https://files.pythonhosted.org/packages/3c/d1/4a5cc64a3cad22821ae4c3f7a90456a08ca19457d8354f4abf46ad03c7e8/lupa-2.8-cp314-cp314t-win_amd64.whl", hash = "sha256:d8766aff03a78c80ad2d188a8bdb216de5ec838359cd87e05bbdfa56394a6105", size = 2209250, upload-time = "2026-04-15T20:07:11.906Z" },
{ url = "https://files.pythonhosted.org/packages/37/7c/cdcb654daf668192aaf36b0aeb94f2281dad092aaa5003688691131736ea/lupa-2.8-cp314-cp314t-win_arm64.whl", hash = "sha256:91d622777febda3ab1bed1d45295f2f32a4680c7b3d7caf8c669998ed5c44118", size = 1126735, upload-time = "2026-04-15T20:07:15.434Z" },
{ url = "https://files.pythonhosted.org/packages/1d/44/de1961ad38e17cd326a53c246c7e3b91178ed578f4cf22ffcd5e7e11b041/lupa-2.8-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:b036738282a5acd2e71fdddb317c9df8b87c1673aa57f403d05fcc2be8abc4ba", size = 1186020, upload-time = "2026-04-15T20:07:35.017Z" },
{ url = "https://files.pythonhosted.org/packages/13/c2/276f0b9dc8bcc5a8a58af5316dfa0e6f56be3613dd6dbcc8d3d2cb6559ba/lupa-2.8-cp39-abi3-manylinux2010_i686.manylinux_2_12_i686.manylinux_2_28_i686.whl", hash = "sha256:ac6b6e8d0e617e26a98cbb44880bcd75de5d32b3ad7b3b3793583909292b47ed", size = 1468944, upload-time = "2026-04-15T20:07:37.782Z" },
{ url = "https://files.pythonhosted.org/packages/63/38/52934e52a5180dc6425d20284d004fe4b27a4f9171a82dc99fb67af250bf/lupa-2.8-cp39-abi3-manylinux2014_armv7l.manylinux_2_17_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:ba3a7dd839f90c3d2e53bebe3c192b1f3f9fd720a6781256405123211fd0dce6", size = 1172998, upload-time = "2026-04-15T20:07:40.812Z" },
{ url = "https://files.pythonhosted.org/packages/c7/82/76b3809bd0839d9b3b4ec58d06591e08f17337b6d9576877cb9d48b34e94/lupa-2.8-cp39-abi3-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:d7edb13a7a5250b5c6c22d1495d9e842b5c9fc5081c8fe6b5efe2112fe3e41f9", size = 1449975, upload-time = "2026-04-15T20:07:44.262Z" },
{ url = "https://files.pythonhosted.org/packages/16/07/2f89d54f747c67c23b4b9ae4aa8c8dd06bb409155dedcf406157f2736b66/lupa-2.8-cp39-abi3-manylinux_2_34_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:891f72e0bffbed1e4175f975aeb2a083956586a100066525e1be485f617f7b25", size = 1281944, upload-time = "2026-04-15T20:07:46.458Z" },
{ url = "https://files.pythonhosted.org/packages/e7/bd/7375d2b0fcae79d806baf52a76f26c96964593f58e1372d13ae5ac09c676/lupa-2.8-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:a295f87b5b7ebbfd5191932e8cb0e51df3c7769101ac6b6c7d7c9fb27bfd1307", size = 1910455, upload-time = "2026-04-15T20:07:49.75Z" },
{ url = "https://files.pythonhosted.org/packages/8b/0c/8abb3bc0e08b311fc01db05b6e9f9ff31a8f65e4fc3f0aeb05cfef75c8ac/lupa-2.8-cp39-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:4fe5d7a810b64ea8511eb885fc8cdde042ee5ff7b7d08ae78f32449756acb177", size = 1155548, upload-time = "2026-04-15T20:07:52.657Z" },
{ url = "https://files.pythonhosted.org/packages/80/2e/9eeecd3f493099721c1d3f31beeca23a4237db1a54223684df4dc96aa1bd/lupa-2.8-cp39-abi3-musllinux_1_2_i686.whl", hash = "sha256:bfc470012ef66ad064c7bd77416af03a3452ef630b04b9012595ea13f2e54518", size = 1489232, upload-time = "2026-04-15T20:07:54.92Z" },
{ url = "https://files.pythonhosted.org/packages/c3/13/731c99dc2e7652ae818a6de45bdf0142049f7cb566049061c898355f1891/lupa-2.8-cp39-abi3-musllinux_1_2_ppc64le.whl", hash = "sha256:250e035fdaffe8c87093e3ebc206ac29a26131b1568ea711d780c26001ce96e7", size = 1466321, upload-time = "2026-04-15T20:07:57.627Z" },
{ url = "https://files.pythonhosted.org/packages/de/71/3ad8cc4fc05a77dc0d3f7079348bd1cad4675a0d14c24f8e6a3ce5f008f7/lupa-2.8-cp39-abi3-musllinux_1_2_riscv64.whl", hash = "sha256:b9bddb09acfffb4f828f790f444b11dc0cca591afea1a244d9329eea2d20c003", size = 1288577, upload-time = "2026-04-15T20:07:59.913Z" },
{ url = "https://files.pythonhosted.org/packages/d8/b2/1175f6d0aa7b68627fbe2f58bd1e8bea36a89d10dfd67671d2b024c96162/lupa-2.8-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:2e64acbbd47e9b82a64405a39e0d2b36a5a7dad8ab41c0f3437f572f7d282ba3", size = 2444866, upload-time = "2026-04-15T20:08:02.753Z" },
{ url = "https://files.pythonhosted.org/packages/92/f7/e78df680c7a0ea452daac07467ca188d63c2c00ca1c884c0a50e27eb83b5/lupa-2.8-pp311-pypy311_pp73-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:32e4e5103bbddcdd2458fb2ccae6c8ba11c9997c711d7e379e0d45551d109c76", size = 1778509, upload-time = "2026-04-15T20:08:21.784Z" },
{ url = "https://files.pythonhosted.org/packages/e6/23/0e53cabb16b2a8aa9cf1fde499c097d8942c5dab709fc8e921f3b824b18b/lupa-2.8-pp311-pypy311_pp73-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:7667001804657496dee9feced2daae5000b4604a3218dd8e6b7b754982ba88b8", size = 2300480, upload-time = "2026-04-15T20:08:24.394Z" },
{ url = "https://files.pythonhosted.org/packages/7e/85/0271227eab939921a12ebba5d17aa4cd18346aa534ca7f5da09cd0b63dd4/lupa-2.8-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:86f6f668966965b15247dc32d064cfe7be67b71e584ccfacbe2f637575296878", size = 1847445, upload-time = "2026-04-15T20:08:27.031Z" },
]
[[package]]
name = "lxml"
version = "6.1.1"