Skip to content

academy.task

SafeTaskExitError

Bases: Exception

Exception that can be raised inside a task to safely exit it.

spawn_guarded_background_task 

spawn_guarded_background_task(
    coro: Coroutine[Any, Any, Any],
    *,
    name: str,
    log_exception: bool = True
) -> Task[Any]

Run a coroutine safely in the background.

Launches the coroutine as an asyncio task. Optionally attaches logging to any exception raised, and/or exiting on any exception raised. Tasks can raise SafeTaskExit to signal the task is finished but should not cause a system exit.

Source: https://stackoverflow.com/questions/62588076

Parameters:

  • coro (Coroutine[Any, Any, Any]) –

    Coroutine to run as task.

  • name (str) –

    name of background task. Unlike asyncio.create_task, this is required

  • log_exception (bool, default: True ) –

    Write exception to the log. Set to false if exceptions are already logged by coro.

Returns:

Source code in academy/task.py 
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
def spawn_guarded_background_task(
    coro: Coroutine[Any, Any, Any],
    *,
    name: str,
    log_exception: bool = True,
) -> asyncio.Task[Any]:
    """Run a coroutine safely in the background.

    Launches the coroutine as an asyncio task. Optionally attaches logging
    to any exception raised, and/or exiting on any exception raised.
    Tasks can raise [`SafeTaskExit`][academy.task.SafeTaskExitError]
    to signal the task is finished but should not cause a system exit.

    Source: https://stackoverflow.com/questions/62588076

    Args:
        coro: Coroutine to run as task.
        name: name of background task. Unlike asyncio.create_task, this is
            required
        log_exception: Write exception to the log. Set to false if exceptions
            are already logged by coro.

    Returns:
        Asyncio task.
    """
    if log_exception:
        fut = asyncio.ensure_future(coro)
        coro = execute_and_log_traceback(fut)

    task = asyncio.create_task(
        coro,
        name=name,
    )
    task.add_done_callback(_exit_on_error)

    return task