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

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?

Solution

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.

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())

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())

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())