From bb973bf01c48ef71203c5f5fb67bdac6bfe81234 Mon Sep 17 00:00:00 2001 From: David Traina <44659830+DavidTraina@users.noreply.github.com> Date: Thu, 5 Nov 2026 00:42:30 -0600 Subject: [PATCH 1/3] feat: add iter_observation_pages helper for observations v2 Since langfuse/langfuse#19043, GET /api/public/v2/observations can return short or empty pages while meta.cursor is set, and a request can take up to 45 s, well past the SDK's 5 s default timeout. Add iter_observation_pages and aiter_observation_pages, which page until the cursor is absent and send a 60 s per-request timeout, so callers don't have to hand-roll the loop. --- langfuse/__init__.py | 4 + langfuse/_client/pagination.py | 109 ++++++++++++++++++++ tests/unit/test_pagination.py | 183 +++++++++++++++++++++++++++++++++ 3 files changed, 296 insertions(+) create mode 100644 langfuse/_client/pagination.py create mode 100644 tests/unit/test_pagination.py diff --git a/langfuse/__init__.py b/langfuse/__init__.py index 222d01079..055100304 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. + `iter_observation_pages` / `aiter_observation_pages` page through observations. Quickstart: @@ -64,6 +65,7 @@ from ._client.constants import ObservationTypeLiteral from ._client.get_client import get_client from ._client.observe import observe +from ._client.pagination import aiter_observation_pages, iter_observation_pages from ._client.propagation import propagate_attributes from ._client.skills import discover_skills from ._client.span import ( @@ -107,6 +109,8 @@ "get_client", "observe", "propagate_attributes", + "iter_observation_pages", + "aiter_observation_pages", "discover_skills", "SkillReference", "SkillsConfig", diff --git a/langfuse/_client/pagination.py b/langfuse/_client/pagination.py new file mode 100644 index 000000000..d7d89a129 --- /dev/null +++ b/langfuse/_client/pagination.py @@ -0,0 +1,109 @@ +"""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 iter_observation_pages( + api: LangfuseAPI, + *, + limit: int = 50, + cursor: Optional[str] = None, + max_items: Optional[int] = None, + request_options: Optional[RequestOptions] = None, + **params: Any, +) -> Iterator[ObservationsV2Response]: + """Yield pages of `api.observations.get_many` 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 loop. 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: + api: `Langfuse.api` or a standalone `LangfuseAPI`. + 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. + request_options: Passed to every request. `timeout_in_seconds` defaults + to 60 because the server may scan for up to 45 s per request. + **params: Other `get_many` arguments, e.g. `trace_id`, `fields`, + `from_start_time`, `filter`. + + 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, iter_observation_pages + + langfuse = get_client() + since = datetime.now(timezone.utc) - timedelta(days=7) + + for page in iter_observation_pages( + langfuse.api, trace_id="abc123", from_start_time=since, limit=1000 + ): + for observation in page.data: + print(observation.id, observation.name) + ``` + """ + options: RequestOptions = { + "timeout_in_seconds": _FETCH_TIMEOUT_SECONDS, + **(request_options or {}), + } + fetched = 0 + while max_items is None or fetched < max_items: + page = api.observations.get_many( + limit=limit if max_items is None else min(limit, max_items - fetched), + cursor=cursor, + request_options=options, + **params, + ) + yield page + fetched += len(page.data) + cursor = page.meta.cursor + if not cursor: + return + + +async def aiter_observation_pages( + api: AsyncLangfuseAPI, + *, + limit: int = 50, + cursor: Optional[str] = None, + max_items: Optional[int] = None, + request_options: Optional[RequestOptions] = None, + **params: Any, +) -> AsyncIterator[ObservationsV2Response]: + """Async version of `iter_observation_pages`; pass `Langfuse.async_api`.""" + options: RequestOptions = { + "timeout_in_seconds": _FETCH_TIMEOUT_SECONDS, + **(request_options or {}), + } + fetched = 0 + while max_items is None or fetched < max_items: + page = await api.observations.get_many( + limit=limit if max_items is None else min(limit, max_items - fetched), + cursor=cursor, + request_options=options, + **params, + ) + yield page + fetched += len(page.data) + 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..f35a0e07f --- /dev/null +++ b/tests/unit/test_pagination.py @@ -0,0 +1,183 @@ +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 import aiter_observation_pages, iter_observation_pages +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 collect( + mode: str, observations: ScriptedObservations, **kwargs: Any +) -> Tuple[List[ObservationsV2Response], Optional[Exception]]: + api: Any = SimpleNamespace(observations=observations) + pages: List[ObservationsV2Response] = [] + + if mode == "sync": + try: + for p in iter_observation_pages(api, **kwargs): + pages.append(p) + except Exception as e: + return pages, e + return pages, None + + async def run() -> Optional[Exception]: + try: + async for p in aiter_observation_pages(api, **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: + api = LangfuseAPI(**client_kwargs, httpx_client=http) + pages = list(iter_observation_pages(api, trace_id="t1")) + else: + + async def run() -> List[ObservationsV2Response]: + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport) as http: + api = AsyncLangfuseAPI(**client_kwargs, httpx_client=http) + return [p async for p in aiter_observation_pages(api, 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.extensions["timeout"]["read"] == 60 for r in requests) From 3f97688d1bae33e2499a089b6232cab2bb880c05 Mon Sep 17 00:00:00 2001 From: David Traina <44659830+DavidTraina@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:07:20 -0600 Subject: [PATCH 2/3] fix: count page rows before yielding A caller that mutates page.data could otherwise make max_items undercount. --- langfuse/_client/pagination.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/langfuse/_client/pagination.py b/langfuse/_client/pagination.py index d7d89a129..6403725a9 100644 --- a/langfuse/_client/pagination.py +++ b/langfuse/_client/pagination.py @@ -73,8 +73,8 @@ def iter_observation_pages( request_options=options, **params, ) - yield page fetched += len(page.data) + yield page cursor = page.meta.cursor if not cursor: return @@ -102,8 +102,8 @@ async def aiter_observation_pages( request_options=options, **params, ) - yield page fetched += len(page.data) + yield page cursor = page.meta.cursor if not cursor: return From e050ae81515cfddc641c6ef569f3ab5ef222099e Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Thu, 5 Nov 2026 09:35:12 +0000 Subject: [PATCH 3/3] refactor: move observation page iteration onto the Langfuse client Expose iter_observation_pages and aiter_observation_pages as Langfuse methods with typed parameters mirroring observations.get_many, instead of top-level functions that take the API client as an argument. The paging loop stays in langfuse/_client/pagination.py. Co-authored-by: Hassieb Pakzad --- langfuse/__init__.py | 5 +- langfuse/_client/client.py | 196 ++++++++++++++++++++++++++++----- langfuse/_client/pagination.py | 82 ++++---------- tests/unit/test_pagination.py | 25 +++-- 4 files changed, 208 insertions(+), 100 deletions(-) diff --git a/langfuse/__init__.py b/langfuse/__init__.py index 055100304..23c77c169 100644 --- a/langfuse/__init__.py +++ b/langfuse/__init__.py @@ -15,7 +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. - `iter_observation_pages` / `aiter_observation_pages` page through observations. + `Langfuse.iter_observation_pages` / `Langfuse.aiter_observation_pages` page through observations. Quickstart: @@ -65,7 +65,6 @@ from ._client.constants import ObservationTypeLiteral from ._client.get_client import get_client from ._client.observe import observe -from ._client.pagination import aiter_observation_pages, iter_observation_pages from ._client.propagation import propagate_attributes from ._client.skills import discover_skills from ._client.span import ( @@ -109,8 +108,6 @@ "get_client", "observe", "propagate_attributes", - "iter_observation_pages", - "aiter_observation_pages", "discover_skills", "SkillReference", "SkillsConfig", 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 index 6403725a9..f7494d971 100644 --- a/langfuse/_client/pagination.py +++ b/langfuse/_client/pagination.py @@ -9,66 +9,28 @@ _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 = 50, - cursor: Optional[str] = None, - max_items: Optional[int] = None, - request_options: Optional[RequestOptions] = None, + limit: int, + cursor: Optional[str], + max_items: Optional[int], + request_options: Optional[RequestOptions], **params: Any, ) -> Iterator[ObservationsV2Response]: - """Yield pages of `api.observations.get_many` 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 loop. 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: - api: `Langfuse.api` or a standalone `LangfuseAPI`. - 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. - request_options: Passed to every request. `timeout_in_seconds` defaults - to 60 because the server may scan for up to 45 s per request. - **params: Other `get_many` arguments, e.g. `trace_id`, `fields`, - `from_start_time`, `filter`. - - 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, iter_observation_pages - - langfuse = get_client() - since = datetime.now(timezone.utc) - timedelta(days=7) - - for page in iter_observation_pages( - langfuse.api, trace_id="abc123", from_start_time=since, limit=1000 - ): - for observation in page.data: - print(observation.id, observation.name) - ``` - """ - options: RequestOptions = { - "timeout_in_seconds": _FETCH_TIMEOUT_SECONDS, - **(request_options or {}), - } + options = _with_fetch_timeout(request_options) fetched = 0 while max_items is None or fetched < max_items: page = api.observations.get_many( - limit=limit if max_items is None else min(limit, max_items - fetched), + limit=_next_limit(limit, max_items, fetched), cursor=cursor, request_options=options, **params, @@ -83,21 +45,17 @@ def iter_observation_pages( async def aiter_observation_pages( api: AsyncLangfuseAPI, *, - limit: int = 50, - cursor: Optional[str] = None, - max_items: Optional[int] = None, - request_options: Optional[RequestOptions] = None, + limit: int, + cursor: Optional[str], + max_items: Optional[int], + request_options: Optional[RequestOptions], **params: Any, ) -> AsyncIterator[ObservationsV2Response]: - """Async version of `iter_observation_pages`; pass `Langfuse.async_api`.""" - options: RequestOptions = { - "timeout_in_seconds": _FETCH_TIMEOUT_SECONDS, - **(request_options or {}), - } + options = _with_fetch_timeout(request_options) fetched = 0 while max_items is None or fetched < max_items: page = await api.observations.get_many( - limit=limit if max_items is None else min(limit, max_items - fetched), + limit=_next_limit(limit, max_items, fetched), cursor=cursor, request_options=options, **params, diff --git a/tests/unit/test_pagination.py b/tests/unit/test_pagination.py index f35a0e07f..f2a1fc898 100644 --- a/tests/unit/test_pagination.py +++ b/tests/unit/test_pagination.py @@ -6,7 +6,7 @@ import httpx import pytest -from langfuse import aiter_observation_pages, iter_observation_pages +from langfuse._client.client import Langfuse from langfuse.api import ( AsyncLangfuseAPI, LangfuseAPI, @@ -59,15 +59,21 @@ def scripted(mode: str, *responses: Any) -> ScriptedObservations: 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: Any = SimpleNamespace(observations=observations) + api = SimpleNamespace(observations=observations) pages: List[ObservationsV2Response] = [] if mode == "sync": try: - for p in iter_observation_pages(api, **kwargs): + for p in client_with(api=api).iter_observation_pages(**kwargs): pages.append(p) except Exception as e: return pages, e @@ -75,7 +81,7 @@ def collect( async def run() -> Optional[Exception]: try: - async for p in aiter_observation_pages(api, **kwargs): + async for p in client_with(async_api=api).aiter_observation_pages(**kwargs): pages.append(p) except Exception as e: return e @@ -166,18 +172,21 @@ def handler(request: httpx.Request) -> httpx.Response: ) if mode == "sync": with httpx.Client(transport=httpx.MockTransport(handler)) as http: - api = LangfuseAPI(**client_kwargs, httpx_client=http) - pages = list(iter_observation_pages(api, trace_id="t1")) + 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: - api = AsyncLangfuseAPI(**client_kwargs, httpx_client=http) - return [p async for p in aiter_observation_pages(api, trace_id="t1")] + 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)