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

14 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 13:38 +0000

1from typing import List, Optional 1a

2 

3from fastapi import Body 1a

4from pydantic import BaseModel 1a

5 

6from prefect.server import models 1a

7from prefect.server.models.task_workers import TaskWorkerResponse 1a

8from prefect.server.utilities.server import PrefectRouter 1a

9 

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

11 

12 

13class TaskWorkerFilter(BaseModel): 1a

14 task_keys: List[str] 1a

15 

16 

17@router.post("/filter") 1a

18async def read_task_workers( 1a

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()