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

1""" 

2Routes for interacting with task 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="/task_run_states", tags=["Task Run States"] 

17) 

18 

19 

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

21async def read_task_run_state( 

22 task_run_state_id: UUID = Path( 

23 ..., description="The task run state id", alias="id" 

24 ), 

25 db: PrefectDBInterface = Depends(provide_database_interface), 

26) -> schemas.states.State: 

27 """ 

28 Get a task run state by id. 

29 

30 For more information, see https://docs.prefect.io/v3/concepts/tasks. 

31 """ 

32 async with db.session_context() as session: 

33 task_run_state = await models.task_run_states.read_task_run_state( 

34 session=session, task_run_state_id=task_run_state_id 

35 ) 

36 if not task_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_code=status.HTTP_404_NOT_FOUND, detail="Flow run state not found" 

39 ) 

40 return task_run_state 

41 

42 

43@router.get("/") 

44async def read_task_run_states( 

45 task_run_id: UUID, 

46 db: PrefectDBInterface = Depends(provide_database_interface), 

47) -> List[schemas.states.State]: 

48 """ 

49 Get states associated with a task run. 

50 """ 

51 async with db.session_context() as session: 

52 return await models.task_run_states.read_task_run_states( 

53 session=session, task_run_id=task_run_id 

54 )