Hi! I'm learning the basics. I have the server up and running, all good on this side. I would like to know what is a preferred way of obtaining the total count of completed flow runs? I'm writing simple CI pipelines and would like to check that all my flows has completed successfully.

hi @Anton Buyskikh - you'd probably want to use the python client for this. the CLI is for visual / interactive terminal based use

In [1]: from prefect import get_client

In [2]: from prefect.client.schemas.filters import FlowRunFilter
   ...:
   ...: async with get_client() as client:
   ...:     all_runs = await client.read_flow_runs()
   ...:     print(len(all_runs))
   ...:     completed_runs = await client.read_flow_runs(flow_run_filter=FlowRunFilter(state=dict(name=dict(any_=["Completed"]))))
   ...:

200

In [3]: len(completed_runs)
Out[3]: 23

In [4]: completed_runs[0]
Out[4]: FlowRun(id=UUID('feb8e111-fb14-4f6e-a851-88271dcf7d24'), name='discreet-parrot', flow_id=UUID('38426bf5-4564-419b-a8e0-a4a5cd1499ab'), state_id=UUID('26f1e240-2944-47c2-9a0c-cbe6b3ee59c1'), deployment_id=None, deployment_version=None, work_queue_name=None, flow_version=None, parameters={}, idempotency_key=None, context={}, empirical_policy=FlowRunPolicy(max_retries=0, retry_delay_seconds=0.0, retries=0, retry_delay=0, pause_keys=set(), resuming=False), tags=[], parent_task_run_id=None, run_count=1, expected_start_time=DateTime(2024, 8, 2, 18, 0, 35, 73539, tzinfo=Timezone('UTC')), next_scheduled_start_time=None, start_time=DateTime(2024, 8, 2, 18, 0, 35, 175934, tzinfo=Timezone('UTC')), end_time=DateTime(2024, 8, 2, 18, 0, 38, 849396, tzinfo=Timezone('UTC')), total_run_time=datetime.timedelta(seconds=3, microseconds=673462), estimated_run_time=datetime.timedelta(seconds=3, microseconds=673462), estimated_start_time_delta=datetime.timedelta(microseconds=102395), auto_scheduled=False, infrastructure_document_id=None, infrastructure_pid=None, created_by=CreatedBy(id=UUID('aedfbb5e-5079-4185-bf7a-31f634bb54f7'), type='USER', display_value='nateprefectio'), work_queue_id=None, work_pool_id=None, work_pool_name=None, state=Completed(message=None, type=COMPLETED, result=UnpersistedResult(type='unpersisted')), job_variables={}, state_type=StateType.COMPLETED, state_name='Completed')

Thank you @Nate this is exactly what I needed 🙂