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

1import sys 

2import typing 

3 

4if sys.version_info >= (3, 10): # pragma: no cover 

5 from typing import ParamSpec 

6else: # pragma: no cover 

7 from typing_extensions import ParamSpec 

8 

9from starlette._utils import is_async_callable 

10from starlette.concurrency import run_in_threadpool 

11 

12P = ParamSpec("P") 

13 

14 

15class BackgroundTask: 

16 def __init__( 

17 self, func: typing.Callable[P, typing.Any], *args: P.args, **kwargs: P.kwargs 

18 ) -> None: 

19 self.func = func 

20 self.args = args 

21 self.kwargs = kwargs 

22 self.is_async = is_async_callable(func) 

23 

24 async def __call__(self) -> None: 

25 if self.is_async: 

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

27 else: 

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

29 

30 

31class BackgroundTasks(BackgroundTask): 

32 def __init__(self, tasks: typing.Optional[typing.Sequence[BackgroundTask]] = None): 

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

34 

35 def add_task( 

36 self, func: typing.Callable[P, typing.Any], *args: P.args, **kwargs: P.kwargs 

37 ) -> None: 

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

39 self.tasks.append(task) 

40 

41 async def __call__(self) -> None: 

42 for task in self.tasks: 

43 await task()