Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/flow_run_states.py: 57%
19 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1"""
2Routes for interacting with flow run state objects.
3"""
5from typing import List 1a
6from uuid import UUID 1a
8from fastapi import Depends, HTTPException, Path, status 1a
10import prefect.server.models as models 1a
11import prefect.server.schemas as schemas 1a
12from prefect.server.database import PrefectDBInterface, provide_database_interface 1a
13from prefect.server.utilities.server import PrefectRouter 1a
15router: PrefectRouter = PrefectRouter( 1a
16 prefix="/flow_run_states", tags=["Flow Run States"]
17)
20@router.get("/{id:uuid}") 1a
21async def read_flow_run_state( 1a
22 flow_run_state_id: UUID = Path(
23 ..., description="The flow run state id", alias="id"
24 ),
25 db: PrefectDBInterface = Depends(provide_database_interface),
26) -> schemas.states.State:
27 """
28 Get a flow run state by id.
30 For more information, see https://docs.prefect.io/v3/concepts/flows#final-state-determination.
31 """
32 async with db.session_context() as session:
33 flow_run_state = await models.flow_run_states.read_flow_run_state(
34 session=session, flow_run_state_id=flow_run_state_id
35 )
36 if not flow_run_state:
37 raise HTTPException(
38 status.HTTP_404_NOT_FOUND, detail="Flow run state not found"
39 )
40 return flow_run_state
43@router.get("/") 1a
44async def read_flow_run_states( 1a
45 flow_run_id: UUID,
46 db: PrefectDBInterface = Depends(provide_database_interface),
47) -> List[schemas.states.State]:
48 """
49 Get states associated with a flow run.
50 """
51 async with db.session_context() as session:
52 return await models.flow_run_states.read_flow_run_states(
53 session=session, flow_run_id=flow_run_id
54 )