Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/flow_run_states.py: 90%

1""" 

2Routes for interacting with flow run state objects. 

3""" 

4 

5from typing import List 

6from uuid import UUID 

7 

8from fastapi import Depends, HTTPException, Path, status 

9 

10import prefect.server.models as models 

11import prefect.server.schemas as schemas 

12from prefect.server.database import PrefectDBInterface, provide_database_interface 

13from prefect.server.utilities.server import PrefectRouter 

14 

15router: PrefectRouter = PrefectRouter( 

16 prefix="/flow_run_states", tags=["Flow Run States"] 

17) 

18 

19 

20@router.get("/{id:uuid}") 

21async def read_flow_run_state( 

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. 

29 

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: 36 ↛ 40line 36 didn't jump to line 40 because the condition on line 36 was always true

37 raise HTTPException( 

38 status.HTTP_404_NOT_FOUND, detail="Flow run state not found" 

39 ) 

40 return flow_run_state 

41 

42 

43@router.get("/") 

44async def read_flow_run_states( 

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 )