diff --git a/langfuse/__init__.py b/langfuse/__init__.py index 222d01079..23c77c169 100644 --- a/langfuse/__init__.py +++ b/langfuse/__init__.py @@ -15,6 +15,7 @@ - **Prompt management**: `Langfuse.get_prompt`, `Langfuse.create_prompt` with client-side caching and version/label control. - **Full REST API**: `Langfuse.api` (sync) / `Langfuse.async_api` (async) clients. + `Langfuse.iter_observation_pages` / `Langfuse.aiter_observation_pages` page through observations. Quickstart: diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index bb3bfa520..5d128567c 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -15,11 +15,14 @@ from time import time_ns from typing import ( Any, + AsyncIterator, Callable, Dict, + Iterator, List, Literal, Optional, + Sequence, Tuple, Type, Union, @@ -40,6 +43,7 @@ ) from packaging.version import Version +from langfuse._client import pagination from langfuse._client.attributes import ( LangfuseOtelSpanAttributes, _serialize, @@ -107,9 +111,12 @@ LangfuseAPI, MapValue, NotFoundError, + ObservationLevel, + ObservationsV2Response, Prompt_Chat, Prompt_Text, ) +from langfuse.api.core import RequestOptions from langfuse.api.core.datetime_utils import serialize_datetime from langfuse.batch_evaluation import ( BatchEvaluationResult, @@ -444,36 +451,15 @@ def api(self) -> LangfuseAPI: user, session, tags and input/output are on its root observation. Read scores with `api.scores.get_many` and experiments with `api.experiments`. - - **Read observation pages until `meta.cursor` is absent.** Pages can be - short or empty before the end. The server scans for up to 45 seconds per - request, so pass `request_options={"timeout_in_seconds": 60}`, and pass - `from_start_time` when you know a lower bound (otherwise the server - scans back to 2020-01-01). A 422 means the request hit a server resource - limit: narrow the time range rather than retrying unchanged. A `startTime` - condition in the JSON `filter` overrides `from_start_time` and - `to_start_time`. + - **Page through observations with `iter_observation_pages`.** Pages can be + short or empty before the end, and a request can take up to 45 seconds. + `iter_observation_pages` / `aiter_observation_pages` follow `meta.cursor` + until it is absent and raise the per-request timeout accordingly. - For large-scale aggregation (usage/cost by model, user, etc.), prefer the v2 Metrics API (`api.metrics.metrics(...)`) over paginating row-level data. - Reading all observations of a trace: - ```python - observations, cursor = [], None - while True: - page = langfuse.api.observations.get_many( - trace_id=trace_id, - from_start_time=since, - cursor=cursor, - request_options={"timeout_in_seconds": 60}, - ) - observations.extend(page.data) - cursor = page.meta.cursor - if cursor is None: - break - ``` - - - See also: `async_api`, + See also: `async_api`, `iter_observation_pages`, https://langfuse.com/docs/api-and-data-platform/features/query-via-sdk (ingestion lag: #ingestion-lag), https://langfuse.com/docs/api-and-data-platform/features/observations-api, @@ -505,6 +491,164 @@ def async_api(self, value: AsyncLangfuseAPI) -> None: self._resources.async_api = value + def iter_observation_pages( + self, + *, + limit: int = 50, + cursor: Optional[str] = None, + max_items: Optional[int] = None, + fields: Optional[str] = None, + expand_metadata: Optional[str] = None, + name: Optional[str] = None, + user_id: Optional[str] = None, + session_id: Optional[str] = None, + type: Optional[str] = None, + trace_id: Optional[str] = None, + level: Optional[ObservationLevel] = None, + parent_observation_id: Optional[str] = None, + is_root_observation: Optional[bool] = None, + environment: Optional[Union[str, Sequence[str]]] = None, + from_start_time: Optional[datetime] = None, + to_start_time: Optional[datetime] = None, + version: Optional[str] = None, + filter: Optional[str] = None, + request_options: Optional[RequestOptions] = None, + ) -> Iterator[ObservationsV2Response]: + """Yield pages of `GET /api/public/v2/observations` until the server returns no cursor. + + A page can be short or empty while more data remains, so only a missing + `meta.cursor` ends the iteration. Every page is yielded, including empty + ones. To resume later, save `page.meta.cursor` after processing a page and + pass it back as `cursor` with the same other arguments. + + Args: + limit: Rows per request, at most 1000. + cursor: Cursor to resume from. + max_items: Stop after this many rows. The last request asks only for the + remaining rows, so the last page's cursor resumes right after them. + fields: Comma-separated field groups to include, e.g. `"basic,usage,model"`. + Defaults to `core` and `basic` on the server. + expand_metadata: Comma-separated metadata keys to return non-truncated. + name: Filter by observation name. + user_id: Filter by user ID. + session_id: Filter by session ID. + type: Filter by observation type, e.g. `"GENERATION"`. + trace_id: Filter by trace ID. + level: Filter by level, e.g. `"ERROR"`. + parent_observation_id: Filter by the physical parent observation ID. + is_root_observation: Filter by whether an observation is a logical root. + environment: Filter by one or more environments. + from_start_time: Only observations starting at or after this datetime. + to_start_time: Only observations starting before this datetime. + version: Filter by version. + filter: JSON array of filter conditions. Takes precedence over the + individual filter arguments; see `api.observations.get_many`. + request_options: Passed to every request. `timeout_in_seconds` defaults + to 60 because the server may scan for up to 45 s per request. + + Errors propagate and end the iteration. A 422 means the server hit a resource + limit, usually its time budget; the same request tends to fail again, so + narrow the time range or add filters instead of retrying it unchanged. + + Without a lower time bound the server searches back only to 2020-01-01. A + `startTime` condition in `filter` makes the server ignore `from_start_time` + and `to_start_time`, so put both bounds in `filter`. Rows are not + deduplicated; dedupe by `id` if you need unique observations. + + Example: + ```python + from datetime import datetime, timedelta, timezone + + from langfuse import get_client + + langfuse = get_client() + since = datetime.now(timezone.utc) - timedelta(days=7) + + for page in langfuse.iter_observation_pages( + trace_id="abc123", from_start_time=since, limit=1000 + ): + for observation in page.data: + print(observation.id, observation.name) + ``` + """ + return pagination.iter_observation_pages( + self.api, + limit=limit, + cursor=cursor, + max_items=max_items, + request_options=request_options, + fields=fields, + expand_metadata=expand_metadata, + name=name, + user_id=user_id, + session_id=session_id, + type=type, + trace_id=trace_id, + level=level, + parent_observation_id=parent_observation_id, + is_root_observation=is_root_observation, + environment=environment, + from_start_time=from_start_time, + to_start_time=to_start_time, + version=version, + filter=filter, + ) + + def aiter_observation_pages( + self, + *, + limit: int = 50, + cursor: Optional[str] = None, + max_items: Optional[int] = None, + fields: Optional[str] = None, + expand_metadata: Optional[str] = None, + name: Optional[str] = None, + user_id: Optional[str] = None, + session_id: Optional[str] = None, + type: Optional[str] = None, + trace_id: Optional[str] = None, + level: Optional[ObservationLevel] = None, + parent_observation_id: Optional[str] = None, + is_root_observation: Optional[bool] = None, + environment: Optional[Union[str, Sequence[str]]] = None, + from_start_time: Optional[datetime] = None, + to_start_time: Optional[datetime] = None, + version: Optional[str] = None, + filter: Optional[str] = None, + request_options: Optional[RequestOptions] = None, + ) -> AsyncIterator[ObservationsV2Response]: + """Async version of `iter_observation_pages`, using `async_api`. + + Example: + ```python + async for page in langfuse.aiter_observation_pages(trace_id="abc123"): + for observation in page.data: + print(observation.id, observation.name) + ``` + """ + return pagination.aiter_observation_pages( + self.async_api, + limit=limit, + cursor=cursor, + max_items=max_items, + request_options=request_options, + fields=fields, + expand_metadata=expand_metadata, + name=name, + user_id=user_id, + session_id=session_id, + type=type, + trace_id=trace_id, + level=level, + parent_observation_id=parent_observation_id, + is_root_observation=is_root_observation, + environment=environment, + from_start_time=from_start_time, + to_start_time=to_start_time, + version=version, + filter=filter, + ) + @overload def start_observation( self, diff --git a/langfuse/_client/pagination.py b/langfuse/_client/pagination.py new file mode 100644 index 000000000..f7494d971 --- /dev/null +++ b/langfuse/_client/pagination.py @@ -0,0 +1,67 @@ +"""Page through `GET /api/public/v2/observations`.""" + +from typing import Any, AsyncIterator, Iterator, Optional + +from langfuse.api import AsyncLangfuseAPI, LangfuseAPI, ObservationsV2Response +from langfuse.api.core import RequestOptions + +# The server scans for up to 45 s per request; the client-wide default is 5 s. +_FETCH_TIMEOUT_SECONDS = 60 + + +def _with_fetch_timeout(request_options: Optional[RequestOptions]) -> RequestOptions: + return {"timeout_in_seconds": _FETCH_TIMEOUT_SECONDS, **(request_options or {})} + + +def _next_limit(limit: int, max_items: Optional[int], fetched: int) -> int: + return limit if max_items is None else min(limit, max_items - fetched) + + +def iter_observation_pages( + api: LangfuseAPI, + *, + limit: int, + cursor: Optional[str], + max_items: Optional[int], + request_options: Optional[RequestOptions], + **params: Any, +) -> Iterator[ObservationsV2Response]: + options = _with_fetch_timeout(request_options) + fetched = 0 + while max_items is None or fetched < max_items: + page = api.observations.get_many( + limit=_next_limit(limit, max_items, fetched), + cursor=cursor, + request_options=options, + **params, + ) + fetched += len(page.data) + yield page + cursor = page.meta.cursor + if not cursor: + return + + +async def aiter_observation_pages( + api: AsyncLangfuseAPI, + *, + limit: int, + cursor: Optional[str], + max_items: Optional[int], + request_options: Optional[RequestOptions], + **params: Any, +) -> AsyncIterator[ObservationsV2Response]: + options = _with_fetch_timeout(request_options) + fetched = 0 + while max_items is None or fetched < max_items: + page = await api.observations.get_many( + limit=_next_limit(limit, max_items, fetched), + cursor=cursor, + request_options=options, + **params, + ) + fetched += len(page.data) + yield page + cursor = page.meta.cursor + if not cursor: + return diff --git a/tests/unit/test_pagination.py b/tests/unit/test_pagination.py new file mode 100644 index 000000000..f2a1fc898 --- /dev/null +++ b/tests/unit/test_pagination.py @@ -0,0 +1,192 @@ +import asyncio +from datetime import datetime, timezone +from types import SimpleNamespace +from typing import Any, Dict, List, Optional, Tuple, Union + +import httpx +import pytest + +from langfuse._client.client import Langfuse +from langfuse.api import ( + AsyncLangfuseAPI, + LangfuseAPI, + ObservationsV2Response, + ObservationV2, +) +from langfuse.api.core import ApiError + +MODES = pytest.mark.parametrize("mode", ["sync", "async"]) + + +def page(ids: str, cursor: Optional[str] = None) -> ObservationsV2Response: + return ObservationsV2Response( + data=[ + ObservationV2( + id=i, + start_time=datetime(2026, 1, 1, tzinfo=timezone.utc), + project_id="p", + type="SPAN", + ) + for i in ids + ], + meta={"cursor": cursor} if cursor else {}, + ) + + +class ScriptedObservations: + def __init__(self, *responses: Union[ObservationsV2Response, Exception]): + self.responses = list(responses) + self.calls: List[Dict[str, Any]] = [] + + def _next(self, kwargs: Dict[str, Any]) -> ObservationsV2Response: + self.calls.append(kwargs) + response = self.responses.pop(0) + if isinstance(response, Exception): + raise response + return response + + def get_many(self, **kwargs: Any) -> ObservationsV2Response: + return self._next(kwargs) + + +class AsyncScriptedObservations(ScriptedObservations): + async def get_many(self, **kwargs: Any) -> ObservationsV2Response: # type: ignore[override] + return self._next(kwargs) + + +def scripted(mode: str, *responses: Any) -> ScriptedObservations: + cls = ScriptedObservations if mode == "sync" else AsyncScriptedObservations + return cls(*responses) + + +def client_with(api: Optional[Any] = None, async_api: Optional[Any] = None) -> Langfuse: + client = object.__new__(Langfuse) + client._resources = SimpleNamespace(api=api, async_api=async_api) # type: ignore[assignment] + return client + + +def collect( + mode: str, observations: ScriptedObservations, **kwargs: Any +) -> Tuple[List[ObservationsV2Response], Optional[Exception]]: + api = SimpleNamespace(observations=observations) + pages: List[ObservationsV2Response] = [] + + if mode == "sync": + try: + for p in client_with(api=api).iter_observation_pages(**kwargs): + pages.append(p) + except Exception as e: + return pages, e + return pages, None + + async def run() -> Optional[Exception]: + try: + async for p in client_with(async_api=api).aiter_observation_pages(**kwargs): + pages.append(p) + except Exception as e: + return e + return None + + return pages, asyncio.run(run()) + + +def ids(pages: List[ObservationsV2Response]) -> List[str]: + return [o.id for p in pages for o in p.data] + + +@MODES +def test_follows_cursor_through_empty_and_short_pages(mode: str) -> None: + observations = scripted( + mode, + page("ab", "c1"), + page("", "c2"), + page("c", "c3"), + page("de"), + ) + + pages, error = collect(mode, observations, limit=2, trace_id="t1") + + assert error is None + assert [[o.id for o in p.data] for p in pages] == [ + ["a", "b"], + [], + ["c"], + ["d", "e"], + ] + assert [c["cursor"] for c in observations.calls] == [None, "c1", "c2", "c3"] + assert all(c["trace_id"] == "t1" and c["limit"] == 2 for c in observations.calls) + + +@MODES +def test_sends_per_request_timeout_and_keeps_caller_options(mode: str) -> None: + observations = scripted(mode, page("a", "c1"), page("b")) + collect(mode, observations, request_options={"max_retries": 0}) + assert [c["request_options"] for c in observations.calls] == [ + {"timeout_in_seconds": 60, "max_retries": 0} + ] * 2 + + observations = scripted(mode, page("a")) + collect(mode, observations, request_options={"timeout_in_seconds": 120}) + assert observations.calls[0]["request_options"] == {"timeout_in_seconds": 120} + + +@MODES +def test_error_propagates_without_retry(mode: str) -> None: + timeout_422 = ApiError( + status_code=422, headers={}, body={"error": "Request timed out"} + ) + observations = scripted(mode, page("a", "c1"), page("", "c2"), timeout_422) + + pages, error = collect(mode, observations) + + assert error is timeout_422 + assert len(observations.calls) == 3 + assert pages[-1].meta.cursor == observations.calls[-1]["cursor"] == "c2" + + +@MODES +def test_max_items_shrinks_the_last_request_so_the_cursor_resumes_after_it( + mode: str, +) -> None: + observations = scripted(mode, page("a", "c1"), page("bcd", "c2"), page("e", "c3")) + + pages, error = collect(mode, observations, limit=3, max_items=5) + + assert error is None + assert ids(pages) == ["a", "b", "c", "d", "e"] + assert [c["limit"] for c in observations.calls] == [3, 3, 1] + assert pages[-1].meta.cursor == "c3" + + +@MODES +def test_generated_client_uses_60s_timeout_over_the_client_default(mode: str) -> None: + bodies = [{"data": [], "meta": {"cursor": "c1"}}, {"data": [], "meta": {}}] + requests: List[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response(200, json=bodies[len(requests) - 1]) + + client_kwargs: Dict[str, Any] = dict( + base_url="http://langfuse.test", username="pk", password="sk", timeout=5 + ) + if mode == "sync": + with httpx.Client(transport=httpx.MockTransport(handler)) as http: + client = client_with(api=LangfuseAPI(**client_kwargs, httpx_client=http)) + pages = list(client.iter_observation_pages(trace_id="t1")) + else: + + async def run() -> List[ObservationsV2Response]: + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport) as http: + client = client_with( + async_api=AsyncLangfuseAPI(**client_kwargs, httpx_client=http) + ) + return [p async for p in client.aiter_observation_pages(trace_id="t1")] + + pages = asyncio.run(run()) + + assert len(pages) == 2 + assert [r.url.params.get("cursor") for r in requests] == [None, "c1"] + assert all(r.url.params.get("traceId") == "t1" for r in requests) + assert all(r.extensions["timeout"]["read"] == 60 for r in requests)