diff --git a/.agents/skills/redis-access/SKILL.md b/.agents/skills/redis-access/SKILL.md index 7f77801623d..5859806081c 100644 --- a/.agents/skills/redis-access/SKILL.md +++ b/.agents/skills/redis-access/SKILL.md @@ -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/.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 `NativeStore` 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 `NativeStore` 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 diff --git a/.agents/skills/redis-access/references/migration-plan.md b/.agents/skills/redis-access/references/migration-plan.md new file mode 100644 index 00000000000..56a8d8afc72 --- /dev/null +++ b/.agents/skills/redis-access/references/migration-plan.md @@ -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 diff --git a/.agents/skills/redis-access/references/pr-sketches.md b/.agents/skills/redis-access/references/pr-sketches.md new file mode 100644 index 00000000000..e636ca0f800 --- /dev/null +++ b/.agents/skills/redis-access/references/pr-sketches.md @@ -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) -> Result; + pub async fn get(&self, key: &str) -> Result, Error>; + pub async fn set_nx(&self, key: &str, value: impl ToSingleRedisArg, ttl: Duration) -> Result; + pub async fn eval(&self, script: &Script, keys: &[&str], args: impl ToRedisArgs) -> Result; + 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; + +pub async fn release(client: &RedisClient, cronjob_id: &str, owner: &str) -> Result; + +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); + +#[pymethods] +impl NativeRedisClient { + #[staticmethod] + fn connect(settings: RedisSettings) -> PyResult; +} + +// pod_lock.rs +#[pyclass(frozen)] +pub struct NativePodLockHandle(Arc); + +#[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>; + fn release<'py>(&self, py: Python<'py>, cronjob_id: String, owner: String) -> PyResult>; +} +``` + +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 diff --git a/.agents/skills/redis-access/references/transport.md b/.agents/skills/redis-access/references/transport.md deleted file mode 100644 index 6d98a0c4135..00000000000 --- a/.agents/skills/redis-access/references/transport.md +++ /dev/null @@ -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 diff --git a/litellm/_service_logger.py b/litellm/_service_logger.py index 75e88b880ac..6317f18d30b 100644 --- a/litellm/_service_logger.py +++ b/litellm/_service_logger.py @@ -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) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 587505e1011..8ba0f2ec664 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -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: diff --git a/litellm/proxy/response_polling/polling_handler.py b/litellm/proxy/response_polling/polling_handler.py index fe224a52cfa..e797858709f 100644 --- a/litellm/proxy/response_polling/polling_handler.py +++ b/litellm/proxy/response_polling/polling_handler.py @@ -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( diff --git a/litellm/redis/AGENTS.md b/litellm/redis/AGENTS.md index 7f035e4b385..2db518f00c8 100644 --- a/litellm/redis/AGENTS.md +++ b/litellm/redis/AGENTS.md @@ -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 diff --git a/litellm/redis/stores/__init__.py b/litellm/redis/stores/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/litellm/redis/stores/response_polling.py b/litellm/redis/stores/response_polling.py new file mode 100644 index 00000000000..df1f6d78736 --- /dev/null +++ b/litellm/redis/stores/response_polling.py @@ -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 diff --git a/litellm/redis/transport/breaker.py b/litellm/redis/transport/breaker.py new file mode 100644 index 00000000000..17de7ec3ce3 --- /dev/null +++ b/litellm/redis/transport/breaker.py @@ -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 diff --git a/litellm/redis/transport/contract.py b/litellm/redis/transport/contract.py index dc9818417c0..1a39e50c502 100644 --- a/litellm/redis/transport/contract.py +++ b/litellm/redis/transport/contract.py @@ -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: ... diff --git a/litellm/redis/transport/events.py b/litellm/redis/transport/events.py new file mode 100644 index 00000000000..c0c362c1edb --- /dev/null +++ b/litellm/redis/transport/events.py @@ -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 + ) diff --git a/litellm/redis/transport/redis_py.py b/litellm/redis/transport/redis_py.py index 0985634ee90..9ddbac14d2b 100644 --- a/litellm/redis/transport/redis_py.py +++ b/litellm/redis/transport/redis_py.py @@ -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)) diff --git a/pyproject.toml b/pyproject.toml index 6e9e75a5b7f..b5c0ff9daf3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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", diff --git a/tests/_support/fake_redis.py b/tests/_support/fake_redis.py index d6f4b5e3791..a12b2ed4212 100644 --- a/tests/_support/fake_redis.py +++ b/tests/_support/fake_redis.py @@ -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()), ) diff --git a/tests/unit/proxy/response_polling/test_response_polling_handler.py b/tests/unit/proxy/response_polling/test_response_polling_handler.py index 087b273548a..9a3a128044a 100644 --- a/tests/unit/proxy/response_polling/test_response_polling_handler.py +++ b/tests/unit/proxy/response_polling/test_response_polling_handler.py @@ -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"] == [] diff --git a/tests/unit/redis/transport/test_redis_py.py b/tests/unit/redis/transport/test_redis_py.py index ca336903662..2c302b6563a 100644 --- a/tests/unit/redis/transport/test_redis_py.py +++ b/tests/unit/redis/transport/test_redis_py.py @@ -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] == [ diff --git a/uv.lock b/uv.lock index 8cd31eb0828..b0eb44e5c8e 100644 --- a/uv.lock +++ b/uv.lock @@ -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"