scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-10 08:28:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 108

test(redis): satisfy final lint and formatting checks

Browse source
This commit is contained in:
Devin AI 2026-10-27 00:00:38 +00:00 • committed by yujonglee
parent f5f5ef5da1
commit 4dba2f785a
11 changed files with 86 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/caching/_redis_capabilities.py
View file

@ -66,7 +66,9 @@ def failure(error: BaseException) -> CacheFailure:
reason: Final = (
"circuit_open"
if isinstance(error, RedisCircuitBreakerOpenError)
else "timeout" if is_redis_timeout_failure(error) else "other"
else "timeout"
if is_redis_timeout_failure(error)
else "other"
)
return CacheFailure(
kind="unavailable", message=f"Cache operation failed ({type(error).__name__}): {error}", reason=reason

5
litellm/caching/redis_cache.py
View file

@ -544,7 +544,9 @@ def log_redis_failure(
logger: logging.Logger, level: int, message: str, exc: BaseException | CacheFailure, with_traceback: bool = False
) -> None:
detail: Final = exc.message if isinstance(exc, CacheFailure) else exc
circuit_open: Final = exc.reason == "circuit_open" if isinstance(exc, CacheFailure) else isinstance(exc, RedisCircuitBreakerOpenError)
circuit_open: Final = (
exc.reason == "circuit_open" if isinstance(exc, CacheFailure) else isinstance(exc, RedisCircuitBreakerOpenError)
)
if circuit_open:
logger.debug("%s: %s", message, detail, stacklevel=2)
return
@ -2332,7 +2334,6 @@ class RedisCache(BaseCache):
if len(increment_list) == 0:
return None
_redis_client: Final = cast( # cast-ok: RedisCluster stubs omit the shared Redis commands used here
"Redis[bytes]", self._get_async_client()
)

10
litellm/proxy/common_utils/auth_cache_invalidation_pubsub.py
View file

@ -45,7 +45,11 @@ _DECODED_JSON: Final = TypeAdapter(object)
def auth_cache_invalidation_channel(redis_cache: "RedisCache") -> str:
return AUTH_CACHE_INVALIDATION_CHANNEL if redis_cache.namespace is None else f"{redis_cache.namespace}:{AUTH_CACHE_INVALIDATION_CHANNEL}"
return (
AUTH_CACHE_INVALIDATION_CHANNEL
if redis_cache.namespace is None
else f"{redis_cache.namespace}:{AUTH_CACHE_INVALIDATION_CHANNEL}"
)
@dataclass(frozen=True, slots=True)
@ -296,7 +300,9 @@ class AuthCacheInvalidationSubscriber:
return CacheFailure("unavailable", str(error))
async def _consume(self, subscription: CacheSubscription) -> CacheFailure:
while not isinstance(message := await subscription.receive(timeout_seconds=_POLL_TIMEOUT_SECONDS), CacheFailure):
while not isinstance(
message := await subscription.receive(timeout_seconds=_POLL_TIMEOUT_SECONDS), CacheFailure
):
if message is None:
continue
self._apply_message(message)

8
litellm/proxy/common_utils/config_sync_pubsub.py
View file

@ -251,7 +251,9 @@ class ConfigSyncSubscriber:
return CacheFailure("unavailable", str(error))
async def _consume(self, subscription: CacheSubscription) -> CacheFailure:
while not isinstance(message := await subscription.receive(timeout_seconds=_POLL_TIMEOUT_SECONDS), CacheFailure):
while not isinstance(
message := await subscription.receive(timeout_seconds=_POLL_TIMEOUT_SECONDS), CacheFailure
):
if message is None:
continue
if (failure := await self._debounce_and_resync(subscription)) is not None:
@ -271,7 +273,9 @@ class ConfigSyncSubscriber:
async def _wait_for_min_resync_interval(self) -> None:
if self._last_resync_at is None:
return
seconds_until_next_resync: Final = self._min_resync_interval_seconds - (self._monotonic() - self._last_resync_at)
seconds_until_next_resync: Final = self._min_resync_interval_seconds - (
self._monotonic() - self._last_resync_at
)
if seconds_until_next_resync <= 0:
return
verbose_proxy_logger.debug(

6
litellm/proxy/db/db_transaction_queue/pod_lock_manager.py
View file

@ -160,8 +160,10 @@ class PodLockManager:
if not isinstance(released, CacheFailure):
return int(released)
log_redis_failure(
verbose_proxy_logger, logging.WARNING,
f"Atomic release failed for lock_key={lock_key}, falling back to GET+DEL", released
verbose_proxy_logger,
logging.WARNING,
f"Atomic release failed for lock_key={lock_key}, falling back to GET+DEL",
released,
)
current_value = await self.redis_cache.async_get_cache(lock_key)

9
litellm/proxy/db/db_transaction_queue/redis_update_buffer.py
View file

@ -561,11 +561,13 @@ class RedisUpdateBuffer:
if isinstance(result, CacheFailure):
verbose_proxy_logger.error(
"Spend tracking - failed to restore uncommitted transactions to Redis. "
"These spend updates are lost. Error: %s", result.message,
"These spend updates are lost. Error: %s",
result.message,
)
return
verbose_proxy_logger.info(
"Spend tracking - restored %d uncommitted transaction set(s) to Redis for retry on next tick.", len(rpush_list)
"Spend tracking - restored %d uncommitted transaction set(s) to Redis for retry on next tick.",
len(rpush_list),
)
@with_service_target(SPEND_QUEUE_TARGET)
@ -582,7 +584,8 @@ class RedisUpdateBuffer:
if isinstance(result, CacheFailure):
verbose_proxy_logger.error(
"Tool discovery - failed to push %d tool rows to Redis, restoring them for the next tick. Error: %s",
len(items), result.message,
len(items),
result.message,
)
for item in items:
tool_discovery_queue.add_update(item)

18
litellm/proxy/hooks/batch_enqueued_tokens.py
View file

@ -187,8 +187,10 @@ class BatchEnqueuedTokenStore:
if not isinstance(result, CacheFailure):
return result
log_redis_failure(
verbose_proxy_logger, logging.WARNING,
"Redis enqueued-token reserve failed, falling back to in-memory", result
verbose_proxy_logger,
logging.WARNING,
"Redis enqueued-token reserve failed, falling back to in-memory",
result,
)
return await self._reserve_in_memory(tokens=tokens, scopes=scopes, span=litellm_parent_otel_span)
@ -325,8 +327,10 @@ class BatchEnqueuedTokenStore:
if not isinstance(result, CacheFailure):
return
log_redis_failure(
verbose_proxy_logger, logging.WARNING,
"Redis enqueued-token reservation save failed, falling back to in-memory", result
verbose_proxy_logger,
logging.WARNING,
"Redis enqueued-token reservation save failed, falling back to in-memory",
result,
)
await self.internal_usage_cache.async_set_cache(
key=self._record_key(batch_id),
@ -371,8 +375,10 @@ class BatchEnqueuedTokenStore:
if not isinstance(result, CacheFailure):
return result
log_redis_failure(
verbose_proxy_logger, logging.WARNING,
"Redis enqueued-token reservation pop failed, falling back to in-memory", result
verbose_proxy_logger,
logging.WARNING,
"Redis enqueued-token reservation pop failed, falling back to in-memory",
result,
)
return None

47
litellm/proxy/hooks/parallel_request_limiter_v3.py
View file

@ -1254,7 +1254,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
if not isinstance(result, CacheFailure):
return result
self._reject_if_rate_limit_unverifiable("atomic counter read", result)
log_redis_failure(verbose_proxy_logger, logging.WARNING, "Atomic counter read failed, using local mirror", result)
log_redis_failure(
verbose_proxy_logger, logging.WARNING, "Atomic counter read failed, using local mirror", result
)
return await self._batch_get_counter_values(keys=keys, parent_otel_span=parent_otel_span, local_only=True)
def _reject_if_rate_limit_unverifiable(self, failed_operation: str, error: CacheFailure) -> None:
@ -1312,8 +1314,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
await self._refund_later_pipelined_groups(key_groups[index + 1 :], operations[index + 1 :])
self._reject_if_rate_limit_unverifiable("window counter increment", result)
log_redis_failure(
verbose_proxy_logger, logging.WARNING,
f"Window counter increment failed for hash tag {hash_tag}", result
verbose_proxy_logger,
logging.WARNING,
f"Window counter increment failed for hash tag {hash_tag}",
result,
)
all_cache_values.extend(
await self.in_memory_cache_sliding_window(
@ -1609,8 +1613,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
if isinstance(result, CacheFailure):
self._reject_if_rate_limit_unverifiable("concurrency acquire", result)
log_redis_failure(
verbose_proxy_logger, logging.WARNING,
"Concurrency acquire failed, falling back to in-memory gauge", result
verbose_proxy_logger,
logging.WARNING,
"Concurrency acquire failed, falling back to in-memory gauge",
result,
)
async with self._check_and_increment_lock:
return await self._acquire_parallel_slots_in_memory(gauges, slot_id, parent_otel_span)
@ -1642,7 +1648,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
if not isinstance(result, CacheFailure):
return tuple(max(0, count) for count in result)
self._reject_if_rate_limit_unverifiable("concurrency count", result)
log_redis_failure(verbose_proxy_logger, logging.WARNING, "Concurrency count failed, using local mirror", result)
log_redis_failure(
verbose_proxy_logger, logging.WARNING, "Concurrency count failed, using local mirror", result
)
return await self._read_local_gauge_counts(gauge_keys, parent_otel_span)
async def _read_local_gauge_counts(
@ -1760,8 +1768,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
await self._mirror_released_parallel_slots(counter_keys, result, parent_otel_span)
return
log_redis_failure(
verbose_proxy_logger, logging.WARNING,
"Concurrency release failed, falling back to in-memory release", result
verbose_proxy_logger,
logging.WARNING,
"Concurrency release failed, falling back to in-memory release",
result,
)
await self._release_parallel_request_slots_in_memory(counter_keys, slot_id, parent_otel_span)
@ -1789,8 +1799,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
result: Final = future.result()
if isinstance(result, CacheFailure):
log_redis_failure(
verbose_proxy_logger, logging.WARNING,
"Concurrency release failed, the slot stays released in memory only", result
verbose_proxy_logger,
logging.WARNING,
"Concurrency release failed, the slot stays released in memory only",
result,
)
redis_cache.async_release_concurrency(
@ -1975,7 +1987,8 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
await self._refund_applied_descriptor_groups(applied)
self._reject_if_rate_limit_unverifiable("window capacity reservation", result)
log_redis_failure(
verbose_proxy_logger, logging.ERROR,
verbose_proxy_logger,
logging.ERROR,
f"Window capacity reservation failed, refunded {len(applied)} prior descriptors and falling back to memory",
result,
)
@ -2033,7 +2046,8 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
await self._refund_applied_descriptor_groups(applied)
self._reject_if_rate_limit_unverifiable("window capacity reservation", failure)
log_redis_failure(
verbose_proxy_logger, logging.ERROR,
verbose_proxy_logger,
logging.ERROR,
f"Window capacity reservation failed, refunded {len(applied)} batched descriptors and falling back to memory",
failure,
)
@ -2135,7 +2149,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
),
CacheFailure,
):
log_redis_failure(verbose_proxy_logger, logging.WARNING, f"Failed to refund {refund.counter_key} on rollback", result)
log_redis_failure(
verbose_proxy_logger, logging.WARNING, f"Failed to refund {refund.counter_key} on rollback", result
)
def _build_atomic_response(
self,
@ -4244,7 +4260,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
if not isinstance(result, CacheFailure):
return
log_redis_failure(
verbose_proxy_logger, logging.WARNING, "TTL preservation failed, falling back to regular pipeline", result
verbose_proxy_logger,
logging.WARNING,
"TTL preservation failed, falling back to regular pipeline",
result,
)
await self.internal_usage_cache.dual_cache.async_increment_cache_pipeline(increment_list=group_operations)

5
tests/unit/caching/test_redis_cluster_cache.py
View file

@ -661,9 +661,10 @@ async def test_cancelled_typed_cluster_read_releases_its_connection_admission()
startup_nodes=[{"host": "typed-cluster-cancellation", "port": 7000}],
connection_gate=ClusterConnectionGate(limit=1, wait_timeout=60),
)
first: Final = asyncio.create_task(cache.async_read_values(keys=("cancelled",)))
with pytest.warns(FutureWarning, match="async_read_values is experimental and may change"):
first: Final = asyncio.create_task(cache.async_read_values(keys=("cancelled",)))
assert await asyncio.wait_for(entered.get(), timeout=1) == "cancelled"
entered_key: Final = await asyncio.wait_for(entered.get(), timeout=1)
assert entered_key == "cancelled"
first.cancel()
with pytest.raises(asyncio.CancelledError):
await first

6
tests/unit/proxy/_experimental/mcp_server/outbound_credentials/test_redis_distributed_lock.py
View file

@ -133,7 +133,8 @@ async def test_lease_cancellation_is_not_translated_to_an_outage(
) -> None:
lock, transport = backend
transport.side_effect = asyncio.CancelledError
with pytest.raises(asyncio.CancelledError):
async def invoke() -> None:
match operation:
case "acquire":
await lock.acquire("k", "tok", 10.0)
@ -144,6 +145,9 @@ async def test_lease_cancellation_is_not_translated_to_an_outage(
case "is_held":
await lock.is_held("k")
with pytest.raises(asyncio.CancelledError):
await invoke()
@pytest.mark.asyncio
async def test_lease_apis_warn_as_experimental_at_the_lock_callsite(

8
tests/unit/proxy/auth/test_login_utils.py
View file

@ -1509,11 +1509,9 @@ async def test_invalid_shared_throttle_replies_do_not_allow_unchecked_logins(
redis, _ = login_redis_backend
throttle: Final = _throttle(user_limit=1, redis_cache=redis)
with patch("redis.asyncio.client.Redis.register_script", return_value=AsyncMock(return_value=reply)):
with pytest.raises(ValueError):
if phase == "attempt":
await throttle.attempt("admin")
else:
await throttle.record_failure("admin")
operation: Final = throttle.attempt if phase == "attempt" else throttle.record_failure
with pytest.raises(ValueError, match="validation error"):
await operation("admin")
assert not throttle.counters.cache_dict
assert not throttle.blocks.cache_dict