Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/task_run_states.py: 90%
19 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +0000
1"""
2Routes for interacting with task 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="/task_run_states", tags=["Task Run States"]
17)
20@router.get("/{id:uuid}") 1a
21async def read_task_run_state( 1a
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.
30 For more information, see https://docs.prefect.io/v3/concepts/tasks.
31 """
32 async with db.session_context() as session: 1b
33 task_run_state = await models.task_run_states.read_task_run_state( 1b
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 true1b
37 raise HTTPException( 1b
38 status_code=status.HTTP_404_NOT_FOUND, detail="Flow run state not found"
39 )
40 return task_run_state
43@router.get("/") 1a
44async def read_task_run_states( 1a
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: 1bc
52 return await models.task_run_states.read_task_run_states( 1bc
53 session=session, task_run_id=task_run_id
54 )