Tasks

Like Starlette and any other Starlette based frameworks, in Esmerald you can define background tasks to run after the returning response.

You can use the Task and Task with any ASGI framework as if it was native.

This can be useful for those operations that need to happen after the request without blocking the client (the client doesn't have to wait to complete) from receiving that same response.

Also you can simply run them as non-blocking operations without relying on any ASGI framework, simple Python background tasks.

backgrounder.Task

Task(func, *args, **kwargs)

Bases: Repr

Task as a single instance can be easily achieved.

Example

from backgrounder import Task

async def send_email_notification(message: str):
    '''
    Sends an email notification
    '''
    send_notification(message)

task = Task(send_email_notification, "message to someone")

PARAMETER	DESCRIPTION
func	Any callable to be executed by in the background. This can be async def of normal blocking def. Example from backgrounder import Task # For blocking callables def send_notification(message: str) -> None: ... task = Task(send_notification, "A notification") await task() # For async callables async def send_notification(message: str) -> None: ... task = Task(send_notification, "A notification") await task() TYPE: Callable[P, Any]
*args	Any arguments of the callable. Example from backgrounder import Task # For blocking callables def send_notification(message: str, email: str) -> None: ... task = Task(send_notification, "A notification", "user@example.com") # For async callables async def send_notification(message: str, email: str) -> None: ... task = Task(send_notification, "A notification", "user@example.com") TYPE: Any DEFAULT: ()
**kwargs	Any kwyword arguments of the callable. Example from typing import Any from backgrounder import Task data = {"message": "A notification", "email": "user@example.com"} # For blocking callables def send_notification(**kwargs: Any) -> None: message = kwargs.pop("message", None) email = kwargs.pop("email", None) task = Task(send_notification, **data) # For async callables async def send_notification(**kwargs: Any) -> None: message = kwargs.pop("message", None) email = kwargs.pop("email", None) task = Task(send_notification, **data) TYPE: Any DEFAULT: {}

Source code in backgrounder/tasks.py

def __init__(
    self,
    func: Annotated[
        Callable[P, Any],
        Doc(
            """
            Any callable to be executed by in the background.
            This can be `async def` of normal blocking `def`.

            **Example**

            ```python
            from backgrounder import Task

            # For blocking callables
            def send_notification(message: str) -> None:
                ...

            task = Task(send_notification, "A notification")
            await task()

            # For async callables
            async def send_notification(message: str) -> None:
                ...

            task = Task(send_notification, "A notification")
            await task()
            ```
            """
        ),
    ],
    *args: Annotated[
        Any,
        Doc(
            """
            Any arguments of the callable.

            **Example**

            ```python
            from backgrounder import Task

            # For blocking callables
            def send_notification(message: str, email: str) -> None:
                ...

            task = Task(send_notification, "A notification", "user@example.com")

            # For async callables
            async def send_notification(message: str, email: str) -> None:
                ...

            task = Task(send_notification, "A notification", "user@example.com")
            ```
            """
        ),
    ],
    **kwargs: Annotated[
        Any,
        Doc(
            """
            Any kwyword arguments of the callable.

            **Example**

            ```python
            from typing import Any

            from backgrounder import Task

            data = {"message": "A notification", "email": "user@example.com"}

            # For blocking callables
            def send_notification(**kwargs: Any) -> None:
                message = kwargs.pop("message", None)
                email = kwargs.pop("email", None)

            task = Task(send_notification, **data)

            # For async callables
            async def send_notification(**kwargs: Any) -> None:
                message = kwargs.pop("message", None)
                email = kwargs.pop("email", None)

            task = Task(send_notification, **data)
            ```
            """
        ),
    ],
) -> None:
    self.func = enforce_async_callable(func)
    self.args = args
    self.kwargs = kwargs

func instance-attribute

func = enforce_async_callable(func)

args instance-attribute

args = args

kwargs instance-attribute

kwargs = kwargs

backgrounder.Tasks

Tasks(tasks=None, as_group=False)

Bases: Task

Alternatively, the Tasks can also be used to be passed in.

Example

from datetime import datetime

from backgrounder import Task, Tasks

async def send_email_notification(message: str):
    '''
    Sends an email notification
    '''
    send_notification(message)


def write_in_file():
    with open("log.txt", mode="w") as log:
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        content = f"Notification sent @ {now}"
        log.write(content)

tasks = Tasks([
    Task(send_email_notification, message="Account created"),
    Task(write_in_file),
])

await tasks()

When as_group is set to True, it will run all the tasks concurrently (as a group)

Example

from datetime import datetime

from backgrounder import Task, Tasks

async def send_email_notification(message: str):
    '''
    Sends an email notification
    '''
    send_notification(message)


def write_in_file():
    with open("log.txt", mode="w") as log:
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        content = f"Notification sent @ {now}"
        log.write(content)

tasks = Tasks([
    Task(send_email_notification, message="Account created"),
    Task(write_in_file),
], as_group=True)

await tasks()

PARAMETER	DESCRIPTION
tasks	A list of tasks to run execute. Example from datetime import datetime from backgrounder import Task, Tasks async def send_email_notification(message: str): ''' Sends an email notification ''' send_notification(message) def write_in_file(): with open("log.txt", mode="w") as log: now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") content = f"Notification sent @ {now}" log.write(content) tasks = Tasks([ Task(send_email_notification, message="Account created"), Task(write_in_file), ]) await tasks() TYPE: Union[Sequence[Task], None] DEFAULT: None
as_group	Boolean flag indicating if the tasks should be run concurrently, in other words, as a group. Example from datetime import datetime from backgrounder import Task, Tasks async def send_email_notification(message: str): ''' Sends an email notification ''' send_notification(message) def write_in_file(): with open("log.txt", mode="w") as log: now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") content = f"Notification sent @ {now}" log.write(content) tasks = Tasks([ Task(send_email_notification, message="Account created"), Task(write_in_file), ], as_group=True) await tasks() TYPE: bool DEFAULT: False

Source code in backgrounder/tasks.py

def __init__(
    self,
    tasks: Annotated[
        Union[Sequence[Task], None],
        Doc(
            """
            A `list` of [tasks](#tasks) to run execute.

            **Example**

            ```python
            from datetime import datetime

            from backgrounder import Task, Tasks

            async def send_email_notification(message: str):
                '''
                Sends an email notification
                '''
                send_notification(message)


            def write_in_file():
                with open("log.txt", mode="w") as log:
                    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    content = f"Notification sent @ {now}"
                    log.write(content)

            tasks = Tasks([
                Task(send_email_notification, message="Account created"),
                Task(write_in_file),
            ])

            await tasks()
            ```
            """
        ),
    ] = None,
    as_group: Annotated[
        bool,
        Doc(
            """
            Boolean flag indicating if the tasks should be run concurrently, in other
            words, as a group.

            **Example**

            ```python
            from datetime import datetime

            from backgrounder import Task, Tasks

            async def send_email_notification(message: str):
                '''
                Sends an email notification
                '''
                send_notification(message)


            def write_in_file():
                with open("log.txt", mode="w") as log:
                    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    content = f"Notification sent @ {now}"
                    log.write(content)

            tasks = Tasks([
                Task(send_email_notification, message="Account created"),
                Task(write_in_file),
            ], as_group=True)

            await tasks()
            ```
            """
        ),
    ] = False,
):
    self.tasks = list(tasks) if tasks else []
    self.as_group = as_group

add_task

add_task(func, *args, **kwargs)

Another way of adding tasks to the Tasks object.

Example

from datetime import datetime

from backgrounder import Task, Tasks

async def send_email_notification(message: str):
    '''
    Sends an email notification
    '''
    send_notification(message)


def write_in_file():
    with open("log.txt", mode="w") as log:
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        content = f"Notification sent @ {now}"
        log.write(content)

tasks = Tasks()
tasks.add_task(send_email_notification, message="Account created")
tasks.add_task(write_in_file)

await tasks()

Or if you want to run them concurrently.

Example

from datetime import datetime

from backgrounder import Task, Tasks

async def send_email_notification(message: str):
    '''
    Sends an email notification
    '''
    send_notification(message)


def write_in_file():
    with open("log.txt", mode="w") as log:
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        content = f"Notification sent @ {now}"
        log.write(content)

tasks = Tasks(as_group=True)
tasks.add_task(send_email_notification, message="Account created")
tasks.add_task(write_in_file)

await tasks()

PARAMETER	DESCRIPTION
func	TYPE: Callable[P, Any]
*args	TYPE: args DEFAULT: ()
**kwargs	TYPE: kwargs DEFAULT: {}

Source code in backgrounder/tasks.py

def add_task(self, func: Callable[P, Any], *args: P.args, **kwargs: P.kwargs) -> None:
    """
    Another way of adding tasks to the `Tasks` object.

    **Example**

    ```python
    from datetime import datetime

    from backgrounder import Task, Tasks

    async def send_email_notification(message: str):
        '''
        Sends an email notification
        '''
        send_notification(message)


    def write_in_file():
        with open("log.txt", mode="w") as log:
            now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            content = f"Notification sent @ {now}"
            log.write(content)

    tasks = Tasks()
    tasks.add_task(send_email_notification, message="Account created")
    tasks.add_task(write_in_file)

    await tasks()
    ```

    Or if you want to run them concurrently.

    **Example**

    ```python
    from datetime import datetime

    from backgrounder import Task, Tasks

    async def send_email_notification(message: str):
        '''
        Sends an email notification
        '''
        send_notification(message)


    def write_in_file():
        with open("log.txt", mode="w") as log:
            now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            content = f"Notification sent @ {now}"
            log.write(content)

    tasks = Tasks(as_group=True)
    tasks.add_task(send_email_notification, message="Account created")
    tasks.add_task(write_in_file)

    await tasks()
    ```
    """
    task = Task(func, *args, **kwargs)
    self.tasks.append(task)