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

1from typing import List, Optional 

2 

3from fastapi import Body 

4from pydantic import BaseModel 

5 

6from prefect.server import models 

7from prefect.server.models.task_workers import TaskWorkerResponse 

8from prefect.server.utilities.server import PrefectRouter 

9 

10router: PrefectRouter = PrefectRouter(prefix="/task_workers", tags=["Task Workers"]) 

11 

12 

13class TaskWorkerFilter(BaseModel): 

14 task_keys: List[str] 

15 

16 

17@router.post("/filter") 

18async def read_task_workers( 

19 task_worker_filter: Optional[TaskWorkerFilter] = Body( 

20 default=None, description="The task worker filter", embed=True 

21 ), 

22) -> List[TaskWorkerResponse]: 

23 """ 

24 Read active task workers. Optionally filter by task keys. 

25 

26 For more information, see https://docs.prefect.io/v3/how-to-guides/workflows/run-background-tasks. 

27 """ 

28 

29 if task_worker_filter and task_worker_filter.task_keys: 

30 return await models.task_workers.get_workers_for_task_keys( 

31 task_keys=task_worker_filter.task_keys, 

32 ) 

33 

34 else: 

35 return await models.task_workers.get_all_workers()