Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/starlette/background.py: 50%

1from __future__ import annotations 

2 

3from collections.abc import Callable, Sequence 

4from typing import Any, ParamSpec 

5 

6from starlette._utils import is_async_callable 

7from starlette.concurrency import run_in_threadpool 

8 

9P = ParamSpec("P") 

10 

11 

12class BackgroundTask: 

13 def __init__(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None: 

14 self.func = func 

15 self.args = args 

16 self.kwargs = kwargs 

17 self.is_async = is_async_callable(func) 

18 

19 async def __call__(self) -> None: 

20 if self.is_async: 

21 await self.func(*self.args, **self.kwargs) 

22 else: 

23 await run_in_threadpool(self.func, *self.args, **self.kwargs) 

24 

25 

26class BackgroundTasks(BackgroundTask): 

27 def __init__(self, tasks: Sequence[BackgroundTask] | None = None): 

28 self.tasks = list(tasks) if tasks else [] 

29 

30 def add_task(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None: 

31 task = BackgroundTask(func, *args, **kwargs) 

32 self.tasks.append(task) 

33 

34 async def __call__(self) -> None: 

35 for task in self.tasks: 

36 await task()