https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I am seeing an API error. I have a fl...
# ask-marvin
k
@Marvin I am seeing an API error. I have a flow 100203 tasks. I am trying to iterate over all of them via the API in chunks with limit/offset to collect telemetry. I see the following error:
Copy code
Client error "422 Unprocessable Entity" for url "<https://api.prefect.cloud/api/accounts/><>/workspaces/<>/task_runs/filter"

Response:
{
    "exception_message": "Invalid request received.",
    "exception_detail": [
        {
            "type": "less_than_equal",
            "loc": [
                "body",
                "offset"
            ],
            "msg": "Input should be less than or equal to 10001",
            "input": 10200,
            "ctx": {
                "le": 10001
            }
        }
    ],
    "request_body": {
        "flows": null,
        "flow_runs": null,
        "task_runs": {
            "operator": "and_",
            "id": null,
            "name": null,
            "tags": null,
            "state": null,
            "start_time": null,
            "end_time": null,
            "subflow_runs": null,
            "flow_run_id": {
                "any_": [
                    "<>"
                ],
                "is_null_": false
            }
        }
    }
}
m
✅ thought for 8.2 seconds
Could you paste the full error message / response body after “I see the following error:” ? A few details would also help narrow this down quickly: - Which API endpoint are you calling? - Are you querying task runs for a flow run, or tasks from deployment/code metadata? - Roughly what 
limit
and 
offset
values trigger it? - Prefect Cloud or self-hosted server? - Prefect version, if you know it If you want, paste:
Copy code
request URL
request payload / query params
full response status + body
With that, I can help determine whether this is: - an API pagination limitation, - a filtering/query issue, - a known backend bug, - or a better endpoint/query pattern for large numbers of task runs.
k
Here is the code:
async def iter_task_run_pages(flow_run_id: UUID, page_size: int = DEFAULT_PAGE_SIZE) -> AsyncIterator[list[TaskRun]]: """ Yield task runs for the given flow run one page at a time. Streaming a page at a time lets callers fold each page into running totals without ever materializing the flow run's full task-run set in memory. """ task_run_filter = TaskRunFilter(flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id])) offset = 0 async with get_client() as client: while True: page = await client.read_task_runs( task_run_filter=task_run_filter, limit=page_size, offset=offset, ) yield page if len(page) < page_size: return offset += page_size
CC @Marvin
m
✅ thought for 4.2 seconds
I only see 
CC
— the actual error text didn’t come through. Please paste the full error in a code block, for example:
Copy code
HTTP status: ...
response body: ...
request: ...
If easier, just paste these: - endpoint URL - request body / params - full response JSON - whether this is Prefect Cloud or self-hosted Once I have that, I can help debug the pagination issue for the ~100k task runs.
k
m
✅ thought for 12.9 seconds
I can’t open Slack links directly from here, and the thread search didn’t find those messages. Please paste the contents here, especially:
Copy code
error response
request code
endpoint
payload / params
If the code is long, the key part is the API call that uses 
limit
/ 
offset
. For example, paste something like:
Copy code
<http://client.post|client.post>("/task_runs/filter", json={
  "flow_run_filter": ...,
  "limit": 200,
  "offset": 100000
})
and the response:
Copy code
status_code=...
{"detail": "..."}
Without the actual error text, I’d just be guessing.
k
@Marvin The flow has millions of tasks Error:
Copy code
Client error "422 Unprocessable Entity" for url "<https://api.prefect.cloud/api/accounts/><>/workspaces/<>/task_runs/filter"

Response:
{
    "exception_message": "Invalid request received.",
    "exception_detail": [
        {
            "type": "less_than_equal",
            "loc": [
                "body",
                "offset"
            ],
            "msg": "Input should be less than or equal to 10001",
            "input": 10200,
            "ctx": {
                "le": 10001
            }
        }
    ],
    "request_body": {
        "flows": null,
        "flow_runs": null,
        "task_runs": {
            "operator": "and_",
            "id": null,
            "name": null,
            "tags": null,
            "state": null,
            "start_time": null,
            "end_time": null,
            "subflow_runs": null,
            "flow_run_id": {
                "any_": [
                    "<>"
                ],
                "is_null_": false
            }
        }
    }
}
Code:
Copy code
async def iter_task_run_pages(flow_run_id: UUID, page_size: int = DEFAULT_PAGE_SIZE) -> AsyncIterator[list[TaskRun]]:
    """
    Yield task runs for the given flow run one page at a time.

    Streaming a page at a time lets callers fold each page into running totals
    without ever materializing the flow run's full task-run set in memory.
    """
    task_run_filter = TaskRunFilter(flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id]))

    offset = 0
    async with get_client() as client:
        while True:
            page = await client.read_task_runs(
                task_run_filter=task_run_filter,
                limit=page_size,
                offset=offset,
            )
            yield page
            if len(page) < page_size:
                return
            offset += page_size
m
✅ thought for 244.5 seconds
That error is telling you that Prefect Cloud caps 
offset
for this endpoint. In your case:
Copy code
"loc": ["body", "offset"]
"msg": "Input should be less than or equal to 10001"
So your current approach will fail once 
offset > 10001
. A couple important notes: - You’re calling 
POST /task_runs/filter
- Your code increments 
offset += page_size
- With 
page_size=200
, you hit the cap around the 52nd page - Since your flow has millions of task runs, offset pagination is not the right strategy here What to do instead Use cursor-like pagination with a stable sort + a moving filter window, instead of large offsets. For task runs, the practical pattern is: 1. choose a stable sort order 2. fetch the first page with 
offset=0
3. remember the last item from that page 4. fetch the next page using a stricter filter (
start_time
, 
end_time
, etc.) so you never need a large offset Because your current filter is already scoped to one 
flow_run_id
, the best next step is usually time-window pagination. Recommended approach: paginate by 
start_time
 You can page through task runs in ascending 
start_time
order, advancing a cursor each time. Something like:
Copy code
from collections.abc import AsyncIterator
from uuid import UUID

from prefect import get_client
from prefect.client.schemas.filters import (
    TaskRunFilter,
    TaskRunFilterFlowRunId,
    TaskRunFilterStartTime,
)
from prefect.client.schemas.sorting import TaskRunSort
from prefect.client.schemas.objects import TaskRun


async def iter_task_run_pages(
    flow_run_id: UUID,
    page_size: int = 200,
) -> AsyncIterator[list[TaskRun]]:
    cursor = None

    async with get_client() as client:
        while True:
            task_run_filter = TaskRunFilter(
                flow_run_id=TaskRunFilterFlowRunId(any_=[flow_run_id]),
                start_time=(
                    TaskRunFilterStartTime(after_=cursor)
                    if cursor is not None
                    else None
                ),
            )

            page = await client.read_task_runs(
                task_run_filter=task_run_filter,
                sort=TaskRunSort.START_TIME_ASC,
                limit=page_size,
                offset=0,
            )

            if not page:
                return

            yield page

            if len(page) < page_size:
                return

            cursor = page[-1].start_time
Important caveat: multiple task runs can share the same 
start_time
, or 
start_time
can be 
None
for some runs. So the pure time-cursor approach can skip or duplicate rows at boundaries unless you account for ties/nulls. Safer options Option 1: paginate by time windows Instead of trying to do exact page-by-page cursoring, query in bounded windows: - 1 hour at a time - 10 minutes at a time - 1 day at a time For example: - fetch all task runs for 
flow_run_id=X
where 
start_time >= T1
and 
start_time < T2
- if you get too many rows in that window, split the window into smaller windows This is usually the most robust strategy for millions of task runs. Option 2: use time cursor + de-dup by task run id If you’re collecting telemetry, you can often tolerate a little overlap: - sort by 
START_TIME_ASC
- query 
start_time >= cursor
- keep a set / persistent record of already-seen task run IDs - advance cursor to the last seen timestamp That avoids misses, at the cost of some duplicates. Example idea:
Copy code
seen_ids = set()
cursor = None
Then process only unseen IDs. Option 3: aggregate instead of enumerating every task run If your telemetry only needs counts / rates / summaries, it may be much better to: - query task runs by time slice - aggregate incrementally - avoid pulling millions of full task run objects into your client Why this is happening
Even though the Python client accepts an 
offset
argument, Cloud validation on this endpoint rejects offsets above 10001. So this is not a bug in your code specifically; it’s a limitation of the API contract you’re hitting. My recommendation for your case Since you have millions of task runs for a single flow run, I’d do one of these: 1. Best for robustness: iterate over 
start_time
windows and subdivide windows that return too many rows 2. Best for simplicity: 
START_TIME_ASC
+ cursor + de-dup by 
task_run.id
3. Best for efficiency if possible: rethink telemetry collection so it doesn’t require enumerating every task run through the API One more thing to watch Your example code yields the page before checking if it’s empty. That means callers may receive an empty page at the end. Usually better to check first:
Copy code
if not page:
    return
yield page
If you want, I can help you rewrite your iterator into a production-safe paginator for millions of task runs, including: - stable pagination - duplicate avoidance - handling 
start_time=None
- adaptive time-window splitting
3 Views
Open in Slack
PreviousNext
Powered by