Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/ui/flow_runs.py: 90%
37 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 02:04 +0000
1from __future__ import annotations
3import datetime
4from typing import TYPE_CHECKING, List
5from uuid import UUID
7import sqlalchemy as sa
8from fastapi import Body, Depends
9from pydantic import Field
11import prefect.server.schemas as schemas
12from prefect.logging import get_logger
13from prefect.server import models
14from prefect.server.database import PrefectDBInterface, provide_database_interface
15from prefect.server.utilities.schemas.bases import PrefectBaseModel
16from prefect.server.utilities.server import PrefectRouter
17from prefect.types import DateTime
19if TYPE_CHECKING: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true
20 import logging
22logger: "logging.Logger" = get_logger("server.api.ui.flow_runs")
24router: PrefectRouter = PrefectRouter(prefix="/ui/flow_runs", tags=["Flow Runs", "UI"])
27class SimpleFlowRun(PrefectBaseModel):
28 id: UUID = Field(default=..., description="The flow run id.")
29 state_type: schemas.states.StateType = Field(
30 default=..., description="The state type."
31 )
32 timestamp: DateTime = Field(
33 default=...,
34 description=(
35 "The start time of the run, or the expected start time "
36 "if it hasn't run yet."
37 ),
38 )
39 duration: datetime.timedelta = Field(
40 default=..., description="The total run time of the run."
41 )
42 lateness: datetime.timedelta = Field(
43 default=..., description="The delay between the expected and actual start time."
44 )
47@router.post("/history")
48async def read_flow_run_history(
49 sort: schemas.sorting.FlowRunSort = Body(
50 schemas.sorting.FlowRunSort.EXPECTED_START_TIME_DESC
51 ),
52 limit: int = Body(1000, le=1000),
53 offset: int = Body(0, ge=0),
54 flows: schemas.filters.FlowFilter = None,
55 flow_runs: schemas.filters.FlowRunFilter = None,
56 task_runs: schemas.filters.TaskRunFilter = None,
57 deployments: schemas.filters.DeploymentFilter = None,
58 work_pools: schemas.filters.WorkPoolFilter = None,
59 db: PrefectDBInterface = Depends(provide_database_interface),
60) -> List[SimpleFlowRun]:
61 columns = [
62 db.FlowRun.id,
63 db.FlowRun.state_type,
64 db.FlowRun.start_time,
65 db.FlowRun.expected_start_time,
66 db.FlowRun.total_run_time,
67 # Although it isn't returned, we need to select
68 # this field in order to compute `estimated_run_time`
69 db.FlowRun.state_timestamp,
70 ]
71 async with db.session_context() as session:
72 result = await models.flow_runs.read_flow_runs(
73 columns=columns,
74 flow_filter=flows,
75 flow_run_filter=flow_runs,
76 task_run_filter=task_runs,
77 deployment_filter=deployments,
78 work_pool_filter=work_pools,
79 sort=sort,
80 limit=limit,
81 offset=offset,
82 session=session,
83 )
84 return [
85 SimpleFlowRun(
86 id=r.id,
87 state_type=r.state_type,
88 timestamp=r.start_time or r.expected_start_time,
89 duration=r.estimated_run_time,
90 lateness=r.estimated_start_time_delta,
91 )
92 for r in result
93 ]
96@router.post("/count-task-runs")
97async def count_task_runs_by_flow_run(
98 flow_run_ids: list[UUID] = Body(default=..., embed=True, max_items=200),
99 db: PrefectDBInterface = Depends(provide_database_interface),
100) -> dict[UUID, int]:
101 """
102 Get task run counts by flow run id.
103 """
104 async with db.session_context() as session:
105 query = (
106 sa.select(
107 db.TaskRun.flow_run_id,
108 sa.func.count(db.TaskRun.id).label("task_run_count"),
109 )
110 .where(
111 sa.and_(
112 db.TaskRun.flow_run_id.in_(flow_run_ids),
113 sa.not_(db.TaskRun.subflow_run.has()),
114 )
115 )
116 .group_by(db.TaskRun.flow_run_id)
117 )
119 results = await session.execute(query)
121 task_run_counts_by_flow_run = {
122 flow_run_id: task_run_count for flow_run_id, task_run_count in results.t
123 }
125 return {
126 flow_run_id: task_run_counts_by_flow_run.get(flow_run_id, 0)
127 for flow_run_id in flow_run_ids
128 }