Coverage for /usr/local/lib/python3.12/site-packages/prefect/server/api/task_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 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:
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:
37 raise HTTPException(
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:
52 return await models.task_run_states.read_task_run_states(
53 session=session, task_run_id=task_run_id
54 )