How do I run a background task with sanic blueprint instance?

Question

I saw the docs and that caused some confusion for me. When working with blueprints, how do I schedule a background task inside my manager class or my service class for that matter. Any of the cases would work but I would prefer if I could schedule a background task from the manager class.

admin = Blueprint("admin", version=6)

async def task_to_run_in_bg(app):
    await asyncio.sleep(2)
    logger.info('Starting background task')
    for item in ['one', 'two', 'three']:
        logger.info(f'Cycle {item}')
        await asyncio.sleep(2)
    logger.info('background task done')

@admin.route("admin/applications", methods=["GET"], name="get_apps")
async def get_all_apps(request: Request):
    """
    API to fetch all applications and paginate them according to request params
    """
    payload = request.args
    size = int(payload.get('size', 10))
    start = int(payload.get('start', 0))
    response = await AdminInfoManager.get_all_applications(size, start)
    app.add_task(task_to_run_in_bg)
    return send_response(response)

This doesn't work and I don't have any idea why ? If I can access the 'app' inside the manager class, that would be for the best?

Answer 1

It is unclear what you mean by "manager" class.

If you mean inside of your route on your admin blueprint, what I think you really want is this:

@admin.route("admin/applications", methods=["GET"], name="get_apps")
async def get_all_apps(request: Request):
    ...
    request.app.add_task(task_to_run_in_bg)
    ...

This way, you do not need to have your app instance in scope. It is preferred to access it as a property on the request object.

If you instead meant you want to use app.add_task from inside AdminInfoManager methods, it is unclear what that object is. However, you have a couple options.

1. You can always just create tasks the normal way:

from asyncio import get_running_loop

class AdminInfoManager:
    async def some_async_func():
        loop = get_running_loop()
        loop.create_task(some_other_async_func())

2. Or, you could inject your app instance:

class AdminInfoManager:
    def __init__(self, app: Sanic):
        self.app = app

    async def some_async_func():
        self.app.add_task(some_other_async_func())

3. Or, you can use Sanic.get_app() to access your app from anywhere after it has been created.

from sanic import Sanic

class AdminInfoManager:
    async def some_async_func():
        app = Sanic.get_app()
        app.add_task(some_other_async_func())