scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 174

fix(proxy): fetch blog posts without blocking the event loop (#51369)

Browse source 
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Yassin Kortam 2026-11-12 10:32:44 -08:00 • committed by GitHub
parent d29ac75af1
commit a982b76a93
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 218 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
litellm/litellm_core_utils/get_blog_posts.py
View file

@ -8,9 +8,11 @@ Disable remote fetching entirely:
export LITELLM_LOCAL_BLOG_POSTS=True
"""
import asyncio
import os
import time
import xml.etree.ElementTree as ET
from collections.abc import Awaitable, Callable
from email.utils import parsedate_to_datetime
from importlib.resources import files
from typing import Final
@ -78,6 +80,7 @@ class GetBlogPosts:
_cached_posts: tuple[BlogPost, ...] | None = None
_last_fetch_time: float = 0.0
_refresh_lock: tuple[asyncio.AbstractEventLoop, asyncio.Lock] | None = None
@staticmethod
def load_local_blog_posts() -> tuple[BlogPost, ...]:
@ -98,7 +101,7 @@ class GetBlogPosts:
return response.text
@staticmethod
def parse_rss_to_posts(xml_text: str, max_posts: int = BLOG_POSTS_MAX) -> tuple[BlogPost, ...]:
def parse_rss_to_posts(xml_text: str | bytes, max_posts: int = BLOG_POSTS_MAX) -> tuple[BlogPost, ...]:
"""
Parse RSS XML into blog posts.
@ -147,14 +150,52 @@ class GetBlogPosts:
url,
str(e),
)
return cls.load_local_blog_posts()
return cls._cache_posts(cls.load_local_blog_posts())
if not cls.validate_blog_posts(posts):
return cls._cache_posts(cls.load_local_blog_posts())
return cls._cache_posts(posts)
@classmethod
def _cache_posts(cls, posts: tuple[BlogPost, ...]) -> tuple[BlogPost, ...]:
cls._cached_posts = posts
cls._last_fetch_time = time.time()
return posts
@classmethod
async def aget_blog_posts(cls, url: str, fetch: Callable[[str], Awaitable[bytes]]) -> tuple[BlogPost, ...]:
if is_offline_mode() or os.getenv("LITELLM_LOCAL_BLOG_POSTS", "").lower() == "true":
return cls.load_local_blog_posts()
cls._cached_posts = posts
cls._last_fetch_time = now
return posts
cached: Final = cls._cached_posts
if cached is not None and (time.time() - cls._last_fetch_time) < BLOG_POSTS_TTL_SECONDS:
return cached
loop: Final = asyncio.get_running_loop()
state: Final = cls._refresh_lock
lock: Final = state[1] if state is not None and state[0] is loop else asyncio.Lock()
cls._refresh_lock = (loop, lock)
async with lock:
refreshed: Final = cls._cached_posts
if refreshed is not None and (time.time() - cls._last_fetch_time) < BLOG_POSTS_TTL_SECONDS:
return refreshed
return await cls._refresh_blog_posts(url, fetch)
@classmethod
async def _refresh_blog_posts(cls, url: str, fetch: Callable[[str], Awaitable[bytes]]) -> tuple[BlogPost, ...]:
try:
xml_text: Final = await fetch(url)
posts: Final = cls.parse_rss_to_posts(xml_text)
except (httpx.HTTPError, ET.ParseError, ValueError) as e:
verbose_logger.warning(
"LiteLLM: Failed to fetch blog posts from %s: %s. Falling back to local backup.", url, str(e)
)
return cls._cache_posts(cls.load_local_blog_posts())
if not cls.validate_blog_posts(posts):
return cls._cache_posts(cls.load_local_blog_posts())
return cls._cache_posts(posts)
def get_blog_posts(url: str) -> tuple[BlogPost, ...]:

5
litellm/proxy/public_endpoints/public_endpoints.py
View file

@ -17,7 +17,6 @@ from litellm._logging import verbose_logger
from litellm.litellm_core_utils.get_blog_posts import (
BlogPostsResponse,
GetBlogPosts,
get_blog_posts,
)
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
from litellm.litellm_core_utils.offline_mode import is_offline_mode
@ -561,7 +560,9 @@ async def get_litellm_blog_posts() -> BlogPostsResponse:
Falls back to the bundled local backup on any failure.
"""
try:
return BlogPostsResponse(posts=get_blog_posts(url=litellm.blog_posts_url))
return BlogPostsResponse(
posts=await GetBlogPosts.aget_blog_posts(url=litellm.blog_posts_url, fetch=_fetch_remote_bytes)
)
except Exception as e:
verbose_logger.warning("LiteLLM: get_litellm_blog_posts endpoint fallback triggered: %s", str(e))
return BlogPostsResponse(posts=GetBlogPosts.load_local_blog_posts())

183
tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py
View file

@ -1,11 +1,16 @@
"""Tests for the /public/litellm_blog_posts endpoint."""
from unittest.mock import patch
import asyncio
import time
from typing import Final
from unittest.mock import AsyncMock, patch
import httpx
import pytest
from fastapi.testclient import TestClient
from litellm.litellm_core_utils.get_blog_posts import BlogPost, GetBlogPosts
from litellm.litellm_core_utils.get_blog_posts import BLOG_POSTS_TTL_SECONDS, BlogPost, GetBlogPosts
from litellm.proxy.public_endpoints.public_endpoints import get_litellm_blog_posts
SAMPLE_POSTS = (
BlogPost(
@ -27,15 +32,17 @@ def client():
app = FastAPI()
app.include_router(router)
return TestClient(app)
GetBlogPosts._cached_posts = None
try:
yield TestClient(app)
finally:
GetBlogPosts._cached_posts = None
def test_get_blog_posts_returns_response_shape(client):
with patch(
"litellm.proxy.public_endpoints.public_endpoints.get_blog_posts",
return_value=SAMPLE_POSTS,
):
response = client.get("/public/litellm_blog_posts")
GetBlogPosts._cached_posts = SAMPLE_POSTS
GetBlogPosts._last_fetch_time = time.time()
response = client.get("/public/litellm_blog_posts")
assert response.status_code == 200
assert response.json() == {
@ -58,11 +65,9 @@ def test_get_blog_posts_returns_every_fetched_post(client):
for i in range(10)
)
with patch(
"litellm.proxy.public_endpoints.public_endpoints.get_blog_posts",
return_value=many_posts,
):
response = client.get("/public/litellm_blog_posts")
GetBlogPosts._cached_posts = many_posts
GetBlogPosts._last_fetch_time = time.time()
response = client.get("/public/litellm_blog_posts")
assert response.status_code == 200
assert [p["title"] for p in response.json()["posts"]] == [f"Post {i}" for i in range(10)]
@ -71,7 +76,7 @@ def test_get_blog_posts_returns_every_fetched_post(client):
def test_get_blog_posts_returns_local_backup_on_failure(client):
"""Endpoint returns the local backup when the fetcher raises."""
with patch(
"litellm.proxy.public_endpoints.public_endpoints.get_blog_posts",
"litellm.proxy.public_endpoints.public_endpoints._fetch_remote_bytes",
side_effect=Exception("fetch failed"),
):
response = client.get("/public/litellm_blog_posts")
@ -79,3 +84,153 @@ def test_get_blog_posts_returns_local_backup_on_failure(client):
assert response.status_code == 200
backup = [post.model_dump(mode="json") for post in GetBlogPosts.load_local_blog_posts()]
assert response.json()["posts"] == backup
@pytest.mark.asyncio
async def test_blog_fetch_allows_concurrent_coroutine_progress() -> None:
progress: Final = asyncio.Event()
async def fetch(*args: object, **kwargs: object) -> httpx.Response:
await asyncio.sleep(0)
await progress.wait()
return httpx.Response(503, request=httpx.Request("GET", "https://example.test/rss"))
async def other_request() -> None:
await asyncio.sleep(0)
progress.set()
with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client") as client_factory:
client_factory.return_value.get = AsyncMock(side_effect=fetch)
with patch("litellm.litellm_core_utils.get_blog_posts.httpx.get", side_effect=httpx.ConnectTimeout("timeout")):
GetBlogPosts._cached_posts = None
try:
task: Final = asyncio.create_task(other_request())
response: Final = await get_litellm_blog_posts()
assert task.done()
assert response.posts == GetBlogPosts.load_local_blog_posts()
finally:
await task
GetBlogPosts._cached_posts = None
@pytest.mark.asyncio
@pytest.mark.parametrize("failure", [httpx.ConnectTimeout("timeout"), "not xml", "<rss><channel /></rss>"])
async def test_blog_fetch_failure_is_cached(failure: Exception | str) -> None:
with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client") as client_factory:
client_factory.return_value.get = AsyncMock(
side_effect=failure if isinstance(failure, Exception) else None,
return_value=httpx.Response(
200,
text=failure if isinstance(failure, str) else "",
request=httpx.Request("GET", "https://example.test"),
),
)
with patch(
"litellm.litellm_core_utils.get_blog_posts.httpx.get", side_effect=httpx.ConnectTimeout("timeout")
) as sync:
GetBlogPosts._cached_posts = None
try:
first: Final = await get_litellm_blog_posts()
second: Final = await get_litellm_blog_posts()
assert first.posts == second.posts == GetBlogPosts.load_local_blog_posts()
assert client_factory.return_value.get.call_count + sync.call_count == 1
with patch(
"litellm.litellm_core_utils.get_blog_posts.time.time",
return_value=GetBlogPosts._last_fetch_time + BLOG_POSTS_TTL_SECONDS,
):
expired: Final = await get_litellm_blog_posts()
assert expired.posts == first.posts
assert client_factory.return_value.get.call_count == 2
finally:
GetBlogPosts._cached_posts = None
@pytest.mark.asyncio
async def test_blog_fetch_success_is_cached() -> None:
rss: Final = (
"<rss><channel><item><title>Async post</title><link>https://example.test/post</link></item></channel></rss>"
)
with patch("litellm.llms.custom_httpx.http_handler.get_async_httpx_client") as client_factory:
client_factory.return_value.get = AsyncMock(
return_value=httpx.Response(200, text=rss, request=httpx.Request("GET", "https://example.test/rss"))
)
GetBlogPosts._cached_posts = None
try:
first: Final = await get_litellm_blog_posts()
second: Final = await get_litellm_blog_posts()
assert (
first.posts
== second.posts
== (BlogPost(title="Async post", description="", date="", url="https://example.test/post"),)
)
client_factory.return_value.get.assert_awaited_once()
finally:
GetBlogPosts._cached_posts = None
@pytest.mark.asyncio
@pytest.mark.parametrize("expired", [False, True])
@pytest.mark.parametrize("fails", [False, True])
async def test_concurrent_blog_refresh_is_shared(expired: bool, fails: bool) -> None:
started: Final = asyncio.Event()
release: Final = asyncio.Event()
rss: Final = (
b"<rss><channel><item><title>Shared</title><link>https://example.test/post</link></item></channel></rss>"
)
async def fetch(url: str) -> bytes:
started.set()
await release.wait()
if fails:
raise httpx.ConnectTimeout("timeout")
return rss
fetcher: Final = AsyncMock(side_effect=fetch)
GetBlogPosts._cached_posts = SAMPLE_POSTS if expired else None
GetBlogPosts._last_fetch_time = 0.0
tasks: Final = tuple(
asyncio.create_task(GetBlogPosts.aget_blog_posts("https://example.test/rss", fetcher)) for _ in range(32)
)
try:
await started.wait()
await asyncio.sleep(0)
release.set()
results: Final = await asyncio.gather(*tasks)
expected: Final = GetBlogPosts.load_local_blog_posts() if fails else GetBlogPosts.parse_rss_to_posts(rss)
assert all(posts == expected for posts in results)
assert fetcher.await_count == 1
finally:
release.set()
await asyncio.gather(*tasks, return_exceptions=True)
GetBlogPosts._cached_posts = None
@pytest.mark.asyncio
async def test_cancelled_blog_refresh_does_not_wedge_waiters() -> None:
started: Final = asyncio.Event()
release: Final = asyncio.Event()
async def fetch(url: str) -> bytes:
started.set()
await release.wait()
return (
b"<rss><channel><item><title>Recovered</title><link>https://example.test/post</link></item></channel></rss>"
)
GetBlogPosts._cached_posts = None
first: Final = asyncio.create_task(GetBlogPosts.aget_blog_posts("https://example.test/rss", fetch))
await started.wait()
second: Final = asyncio.create_task(GetBlogPosts.aget_blog_posts("https://example.test/rss", fetch))
try:
await asyncio.sleep(0)
first.cancel()
with pytest.raises(asyncio.CancelledError):
await first
release.set()
posts: Final = await asyncio.wait_for(second, timeout=1)
assert posts == (BlogPost(title="Recovered", description="", date="", url="https://example.test/post"),)
finally:
first.cancel()
release.set()
await asyncio.gather(first, second, return_exceptions=True)
GetBlogPosts._cached_posts = None