1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
from typing import Any, Callable, Iterable
from anyio import create_task_group
from typing_extensions import ParamSpec
from litestar.utils.sync import ensure_async_callable
__all__ = ("BackgroundTask", "BackgroundTasks")
P = ParamSpec("P")
class BackgroundTask:
"""A container for a 'background' task function.
Background tasks are called once a Response finishes.
"""
__slots__ = ("fn", "args", "kwargs")
def __init__(self, fn: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None:
"""Initialize ``BackgroundTask``.
Args:
fn: A sync or async function to call as the background task.
*args: Args to pass to the func.
**kwargs: Kwargs to pass to the func
"""
self.fn = ensure_async_callable(fn)
self.args = args
self.kwargs = kwargs
async def __call__(self) -> None:
"""Call the wrapped function with the passed in arguments.
Returns:
None
"""
await self.fn(*self.args, **self.kwargs)
class BackgroundTasks:
"""A container for multiple 'background' task functions.
Background tasks are called once a Response finishes.
"""
__slots__ = ("tasks", "run_in_task_group")
def __init__(self, tasks: Iterable[BackgroundTask], run_in_task_group: bool = False) -> None:
"""Initialize ``BackgroundTasks``.
Args:
tasks: An iterable of :class:`BackgroundTask <.background_tasks.BackgroundTask>` instances.
run_in_task_group: If you set this value to ``True`` than the tasks will run concurrently, using
a :class:`TaskGroup <anyio.abc.TaskGroup>`. Note: This will not preserve execution order.
"""
self.tasks = tasks
self.run_in_task_group = run_in_task_group
async def __call__(self) -> None:
"""Call the wrapped background tasks.
Returns:
None
"""
if self.run_in_task_group:
async with create_task_group() as task_group:
for task in self.tasks:
task_group.start_soon(task)
else:
for task in self.tasks:
await task()
|
