Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/ui_crud_endpoints/latest_release_endpoints.py: 87%

1import asyncio 

2import re 

3from collections import Counter 

4from collections.abc import Awaitable, Mapping 

5from dataclasses import dataclass 

6from types import MappingProxyType 

7from typing import Annotated, Final, Literal, Protocol, TypeAlias 

8 

9import httpx 

10from fastapi import APIRouter, Depends 

11from pydantic import BaseModel, ValidationError 

12 

13from litellm._logging import verbose_proxy_logger 

14from litellm.caching.in_memory_cache import InMemoryCache 

15from litellm.proxy.auth.user_api_key_auth import user_api_key_auth 

16 

17router: Final = APIRouter() 

18 

19LATEST_RELEASE_URL: Final = "https://api.github.com/repos/BerriAI/litellm/releases/latest" 

20LATEST_RELEASE_FETCH_TIMEOUT_SECONDS: Final = 5 

21LATEST_RELEASE_CACHE_TTL_SECONDS: Final = 60 * 60 

22LATEST_RELEASE_UNAVAILABLE_CACHE_TTL_SECONDS: Final = 5 * 60 

23LATEST_RELEASE_CACHE_KEY: Final = "latest_release_info" 

24 

25_RELEASE_BULLET_PATTERN: Final = re.compile(r"^\*\s+(?:([A-Za-z]+)(?:\([^)]*\))?!?:\s)?\S") 

26_NEW_CONTRIBUTOR_PATTERN: Final = re.compile(r"^\*\s+@\S+ made their first contribution\b") 

27 

28_Bucket: TypeAlias = Literal["new_features", "bug_fixes", "other_updates"] 

29_PREFIX_BUCKETS: Final[Mapping[str, _Bucket]] = MappingProxyType({"feat": "new_features", "fix": "bug_fixes"}) 

30 

31 

32class LatestReleaseInfo(BaseModel): 

33 version: str 

34 new_features: int 

35 bug_fixes: int 

36 other_updates: int 

37 release_url: str 

38 

39 

40@dataclass(frozen=True, slots=True) 

41class LatestReleaseUnavailable: 

42 reason: str 

43 

44 

45class _GitHubRelease(BaseModel): 

46 tag_name: str 

47 html_url: str 

48 body: str 

49 

50 

51class _AsyncGetClient(Protocol): 

52 def get(self, url: str, *, timeout: float | None = None) -> Awaitable[httpx.Response]: ... 52 ↛ exitline 52 didn't return from function 'get' because

53 

54 

55_latest_release_cache: Final = InMemoryCache(max_size_in_memory=1, default_ttl=LATEST_RELEASE_CACHE_TTL_SECONDS) 

56_latest_release_fetch_lock: Final = asyncio.Lock() 

57 

58 

59def _default_client() -> _AsyncGetClient: 

60 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client 

61 from litellm.types.llms.custom_http import httpxSpecialProvider 

62 

63 return get_async_httpx_client(llm_provider=httpxSpecialProvider.UI) 

64 

65 

66def _default_cache() -> InMemoryCache: 

67 return _latest_release_cache 

68 

69 

70def _default_fetch_lock() -> asyncio.Lock: 

71 return _latest_release_fetch_lock 

72 

73 

74def _bucket_for(line: str) -> _Bucket | None: 

75 if _NEW_CONTRIBUTOR_PATTERN.match(line) is not None: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 return None 

77 match: Final = _RELEASE_BULLET_PATTERN.match(line) 

78 if match is None: 

79 return None 

80 prefix: Final = match.group(1) 

81 return "other_updates" if prefix is None else _PREFIX_BUCKETS.get(prefix.lower(), "other_updates") 

82 

83 

84def count_release_bullets(body: str) -> Mapping[_Bucket, int]: 

85 """Bucket release-note bullets by conventional-commit type or ``other_updates``.""" 

86 return MappingProxyType(Counter(bucket for line in body.splitlines() if (bucket := _bucket_for(line)) is not None)) 

87 

88 

89def parse_latest_release(response: httpx.Response) -> LatestReleaseInfo | LatestReleaseUnavailable: 

90 if response.status_code != 200: 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true

91 return LatestReleaseUnavailable(reason=f"GitHub responded with status {response.status_code}") 

92 try: 

93 release: Final = _GitHubRelease.model_validate_json(response.content) 

94 except ValidationError as e: 

95 return LatestReleaseUnavailable(reason=f"GitHub release payload was not the expected shape: {e}") 

96 counts: Final = count_release_bullets(release.body) 

97 return LatestReleaseInfo( 

98 version=release.tag_name.removeprefix("v"), 

99 new_features=counts.get("new_features", 0), 

100 bug_fixes=counts.get("bug_fixes", 0), 

101 other_updates=counts.get("other_updates", 0), 

102 release_url=release.html_url, 

103 ) 

104 

105 

106async def fetch_latest_release(client: _AsyncGetClient) -> LatestReleaseInfo | LatestReleaseUnavailable: 

107 try: 

108 response: Final = await client.get(LATEST_RELEASE_URL, timeout=LATEST_RELEASE_FETCH_TIMEOUT_SECONDS) 

109 except httpx.HTTPError as e: 

110 return LatestReleaseUnavailable(reason=f"{type(e).__name__}: {e}") 

111 return parse_latest_release(response) 

112 

113 

114async def get_latest_release_info( 

115 client: _AsyncGetClient, cache: InMemoryCache, fetch_lock: asyncio.Lock 

116) -> LatestReleaseInfo | LatestReleaseUnavailable: 

117 cached: Final = cache.get_cache(LATEST_RELEASE_CACHE_KEY) 

118 if isinstance(cached, (LatestReleaseInfo, LatestReleaseUnavailable)): 

119 return cached 

120 async with fetch_lock: 

121 cached_after_lock: Final = cache.get_cache(LATEST_RELEASE_CACHE_KEY) 

122 if isinstance(cached_after_lock, (LatestReleaseInfo, LatestReleaseUnavailable)): 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true

123 return cached_after_lock 

124 result: Final = await fetch_latest_release(client) 

125 ttl: Final = ( 

126 LATEST_RELEASE_UNAVAILABLE_CACHE_TTL_SECONDS 

127 if isinstance(result, LatestReleaseUnavailable) 

128 else LATEST_RELEASE_CACHE_TTL_SECONDS 

129 ) 

130 cache.set_cache(LATEST_RELEASE_CACHE_KEY, result, ttl=ttl) 

131 return result 

132 

133 

134@router.get( 

135 "/get/latest_release_info", 

136 tags=["UI Settings"], # mutable-ok: FastAPI's route decorator only accepts a list 

137 dependencies=[Depends(user_api_key_auth)], # mutable-ok: FastAPI's route decorator only accepts a list 

138 response_model=LatestReleaseInfo | None, 

139) 

140async def latest_release_info( 

141 client: Annotated[_AsyncGetClient, Depends(_default_client)], 

142 cache: Annotated[InMemoryCache, Depends(_default_cache)], 

143 fetch_lock: Annotated[asyncio.Lock, Depends(_default_fetch_lock)], 

144) -> LatestReleaseInfo | None: 

145 """ 

146 Latest stable LiteLLM GitHub release with its PR count split into new features, bug fixes and other updates. 

147 Returns null when GitHub can't be reached so the dashboard upgrade banner simply doesn't render. 

148 """ 

149 result: Final = await get_latest_release_info(client=client, cache=cache, fetch_lock=fetch_lock) 

150 if isinstance(result, LatestReleaseUnavailable): 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 verbose_proxy_logger.warning("LiteLLM: latest release info unavailable: %s", result.reason) 

152 return None 

153 return result