diff --git a/litellm/litellm_core_utils/get_blog_posts.py b/litellm/litellm_core_utils/get_blog_posts.py
index c3b4114d5b9..1201a73d6ce 100644
--- a/litellm/litellm_core_utils/get_blog_posts.py
+++ b/litellm/litellm_core_utils/get_blog_posts.py
@@ -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, ...]:
diff --git a/litellm/proxy/public_endpoints/public_endpoints.py b/litellm/proxy/public_endpoints/public_endpoints.py
index 45abd3c4d5f..48898afd8fa 100644
--- a/litellm/proxy/public_endpoints/public_endpoints.py
+++ b/litellm/proxy/public_endpoints/public_endpoints.py
@@ -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())
diff --git a/tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py b/tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py
index 5d460617dda..85391bb521b 100644
--- a/tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py
+++ b/tests/unit/proxy/public_endpoints/test_blog_posts_endpoint.py
@@ -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", ""])
+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 = (
+ "- Async posthttps://example.test/post
"
+ )
+ 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"- Sharedhttps://example.test/post
"
+ )
+
+ 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"- Recoveredhttps://example.test/post
"
+ )
+
+ 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