Compare commits

...
Author SHA1 Message Date
Andrew Maguire 52d6cf0ea0 style: apply ruff formatting 2026-02-06 18:09:50 +00:00
Andrew Maguire d110b58e31 chore: bump version to 7.8.4 2026-02-06 18:06:54 +00:00
Andrew Maguire 7d74a967e6 feat(prompts): add capture_errors option to report fetch failures to error tracking
When capture_errors=True and a PostHog client is provided, prompt fetch
failures are reported via client.capture_exception(). This surfaces
prompt API errors in PostHog error tracking without affecting the
existing fallback behavior (stale cache → fallback → raise).
2026-02-06 18:04:28 +00:00
3 changed files with 135 additions and 1 deletions
+19
View File
@@ -56,6 +56,9 @@ class Prompts:
# Or with direct options (no PostHog client needed)
prompts = Prompts(personal_api_key='phx_xxx', host='https://us.posthog.com')
# With error tracking: prompt fetch failures are reported to PostHog
prompts = Prompts(posthog, capture_errors=True)
# Fetch with caching and fallback
template = prompts.get('support-system-prompt', fallback='You are a helpful assistant.')
@@ -74,6 +77,7 @@ class Prompts:
personal_api_key: Optional[str] = None,
host: Optional[str] = None,
default_cache_ttl_seconds: Optional[int] = None,
capture_errors: bool = False,
):
"""
Initialize Prompts.
@@ -83,11 +87,15 @@ class Prompts:
personal_api_key: Direct API key (optional if posthog provided)
host: PostHog host (defaults to app endpoint)
default_cache_ttl_seconds: Default cache TTL (defaults to 300)
capture_errors: If True and a PostHog client is provided, prompt fetch
failures are reported to PostHog error tracking via capture_exception().
"""
self._default_cache_ttl_seconds = (
default_cache_ttl_seconds or DEFAULT_CACHE_TTL_SECONDS
)
self._cache: Dict[str, CachedPrompt] = {}
self._client = posthog if posthog is not None else None
self._capture_errors = capture_errors
if posthog is not None:
self._personal_api_key = getattr(posthog, "personal_api_key", None) or ""
@@ -152,6 +160,8 @@ class Prompts:
return prompt
except Exception as error:
self._maybe_capture_error(error)
# Fallback order:
# 1. Return stale cache (with warning)
if cached is not None:
@@ -211,6 +221,15 @@ class Prompts:
else:
self._cache.clear()
def _maybe_capture_error(self, error: Exception) -> None:
"""Report a prompt fetch error to PostHog error tracking if enabled."""
if not self._capture_errors or self._client is None:
return
try:
self._client.capture_exception(error)
except Exception:
log.debug("[PostHog Prompts] Failed to capture exception to error tracking")
def _fetch_prompt_from_api(self, name: str) -> str:
"""
Fetch prompt from PostHog API.
+115
View File
@@ -509,6 +509,121 @@ class TestPromptsCompile(TestPrompts):
self.assertEqual(result, "Company: Acme")
class TestPromptsCaptureErrors(TestPrompts):
"""Tests for the capture_errors option."""
@patch("posthog.ai.prompts._get_session")
def test_capture_exception_called_on_fetch_failure_with_fallback(
self, mock_get_session
):
"""Should call capture_exception on fetch failure when capture_errors=True."""
mock_get = mock_get_session.return_value.get
mock_get.side_effect = Exception("Network error")
posthog = self.create_mock_posthog()
prompts = Prompts(posthog, capture_errors=True)
result = prompts.get("test-prompt", fallback="fallback prompt")
self.assertEqual(result, "fallback prompt")
posthog.capture_exception.assert_called_once()
captured_exc = posthog.capture_exception.call_args[0][0]
self.assertIn("Network error", str(captured_exc))
@patch("posthog.ai.prompts._get_session")
@patch("posthog.ai.prompts.time.time")
def test_capture_exception_called_on_fetch_failure_with_stale_cache(
self, mock_time, mock_get_session
):
"""Should call capture_exception when falling back to stale cache."""
mock_get = mock_get_session.return_value.get
mock_get.side_effect = [
MockResponse(json_data=self.mock_prompt_response),
Exception("Network error"),
]
mock_time.return_value = 1000.0
posthog = self.create_mock_posthog()
prompts = Prompts(posthog, capture_errors=True)
# First call populates cache
prompts.get("test-prompt", cache_ttl_seconds=60)
# Expire cache
mock_time.return_value = 1061.0
# Second call falls back to stale cache
result = prompts.get("test-prompt", cache_ttl_seconds=60)
self.assertEqual(result, self.mock_prompt_response["prompt"])
posthog.capture_exception.assert_called_once()
@patch("posthog.ai.prompts._get_session")
def test_capture_exception_called_when_error_is_raised(self, mock_get_session):
"""Should call capture_exception even when the error is re-raised (no fallback, no cache)."""
mock_get = mock_get_session.return_value.get
mock_get.side_effect = Exception("Network error")
posthog = self.create_mock_posthog()
prompts = Prompts(posthog, capture_errors=True)
with self.assertRaises(Exception):
prompts.get("test-prompt")
posthog.capture_exception.assert_called_once()
@patch("posthog.ai.prompts._get_session")
def test_no_capture_exception_when_capture_errors_is_false(self, mock_get_session):
"""Should NOT call capture_exception when capture_errors=False (default)."""
mock_get = mock_get_session.return_value.get
mock_get.side_effect = Exception("Network error")
posthog = self.create_mock_posthog()
prompts = Prompts(posthog)
prompts.get("test-prompt", fallback="fallback prompt")
posthog.capture_exception.assert_not_called()
@patch("posthog.ai.prompts._get_session")
def test_no_capture_exception_without_client(self, mock_get_session):
"""Should not error when capture_errors=True but no client provided."""
mock_get = mock_get_session.return_value.get
mock_get.side_effect = Exception("Network error")
prompts = Prompts(personal_api_key="phx_test_key", capture_errors=True)
result = prompts.get("test-prompt", fallback="fallback prompt")
self.assertEqual(result, "fallback prompt")
@patch("posthog.ai.prompts._get_session")
def test_no_capture_exception_on_successful_fetch(self, mock_get_session):
"""Should NOT call capture_exception on successful fetch."""
mock_get = mock_get_session.return_value.get
mock_get.return_value = MockResponse(json_data=self.mock_prompt_response)
posthog = self.create_mock_posthog()
prompts = Prompts(posthog, capture_errors=True)
prompts.get("test-prompt")
posthog.capture_exception.assert_not_called()
@patch("posthog.ai.prompts._get_session")
def test_capture_exception_failure_does_not_affect_fallback(self, mock_get_session):
"""If capture_exception itself throws, the fallback should still be returned."""
mock_get = mock_get_session.return_value.get
mock_get.side_effect = Exception("Network error")
posthog = self.create_mock_posthog()
posthog.capture_exception.side_effect = Exception("capture failed")
prompts = Prompts(posthog, capture_errors=True)
result = prompts.get("test-prompt", fallback="fallback prompt")
self.assertEqual(result, "fallback prompt")
class TestPromptsClearCache(TestPrompts):
"""Tests for the Prompts.clear_cache() method."""
+1 -1
View File
@@ -1,4 +1,4 @@
VERSION = "7.8.3"
VERSION = "7.8.4"
if __name__ == "__main__":
print(VERSION, end="") # noqa: T201