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

fix: type registered health lock script

Browse source
This commit is contained in:
Charan Rathore 2026-09-28 10:31:17 +05:30
parent f2af3de3b3
commit 343360d7af
1 changed files with 6 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
litellm/proxy/health_check_utils/shared_health_check_manager.py
View file

@ -1,8 +1,8 @@
import asyncio
import json
import time
from collections.abc import Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final
from collections.abc import Awaitable, Callable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, cast
from litellm._logging import verbose_proxy_logger
from litellm.caching.redis_cache import RedisCache
@ -38,7 +38,7 @@ class SharedHealthCheckManager:
self.health_check_ttl = health_check_ttl
self.lock_ttl = lock_ttl
self.pod_id = f"pod_{int(time.time() * 1000)}"
self._release_lock_script: Any | None = None
self._release_lock_script: Callable[..., Awaitable[int]] | None = None
@staticmethod
def get_health_check_lock_key() -> str:
@ -111,7 +111,9 @@ end
try:
if self._release_lock_script is None:
self._release_lock_script = script_register(self._COMPARE_AND_DELETE_LOCK_SCRIPT)
self._release_lock_script = cast(
Callable[..., Awaitable[int]], script_register(self._COMPARE_AND_DELETE_LOCK_SCRIPT)
)
result: Final = await self._release_lock_script(keys=[lock_key], args=[json.dumps(self.pod_id)])
if int(result or 0) == 1:
verbose_proxy_logger.info("Pod %s released health check lock", self.pod_id)