How to use filters in the Prefect 2 API

Here is an example of how to use filters when calling the Prefect 2 REST API from the Python client. This specific example calls read_flow_runs for the current flow’s flow_id in order to log the flow run number.

import asyncio
​
from prefect import task, flow, get_run_logger
from prefect.client import get_client
from prefect.context import get_run_context
from prefect.orion.schemas.filters import FlowFilter, FlowFilterId
​
​
@flow
async def get_count_flow_runs():
    logger = get_run_logger()
    context = get_run_context().flow_run.dict()
    flow_id = context['flow_id']
​
    client = get_client()
    flow_runs = await client.read_flow_runs(
        flow_filter=FlowFilter(id=FlowFilterId(any_=[flow_id]))
    )
​
    logger.info(f"This is run number {len(flow_runs)} of flow {flow_id}")
​
​
if __name__ == "__main__":
    asyncio.run(get_count_flow_runs())
2 Likes