Version 1

This documentation refers to the version 1.x of the web framework. The current version of the main branch is documented here.

Background tasks

This page describes how to start background tasks in request handlers, and how to configure background tasks that run periodically during the application's lifetime.

How to handle a request in background

The following example shows how to handle a web request in background, which is the use case for the HTTP 202 Accepted response status code.

import asyncio
from blacksheep import Application, Response, accepted

app = Application(show_error_details=True)
get = app.router.get


async def background_work():
    # simulate a delay...
    await asyncio.sleep(1)
    print("Done!")


@get("/")
def home() -> Response:
    # start a task in background
    asyncio.create_task(background_work())

    # note: the server returns a response immediately, so before the conclusion
    # of the background task
    return accepted("Operation accepted")

How to configure background tasks

The following example shows how to configure a background tasks, including the activation of a service resolved by the DI container, running periodically once every second:

import asyncio
from datetime import datetime

from blacksheep import Application

app = Application()


@app.route("/")
def home():
    return f"Hello, World! {datetime.utcnow().isoformat()}"


def get_current_timestamp():
    return datetime.utcnow().isoformat()


class Foo:
    def __init__(self) -> None:
        pass


async def task_example(app: Application) -> None:
    # example background task, running once every second,
    # this example also shows how to activate a service using the CI container
    while True:
        print(get_current_timestamp())

        my_foo = app.service_provider.get(Foo)
        assert isinstance(my_foo, Foo)
        print("Foo id: ", id(my_foo))

        await asyncio.sleep(1)


async def configure_background_tasks(app):
    asyncio.get_event_loop().create_task(task_example(app))


app.on_start += configure_background_tasks

app.services.add_exact_scoped(Foo)

Last modified on: 2022-02-16 07:52:42

RV