如何在Sanic Blueprint实例中调度后台任务？含管理器类需求

原代码问题分析 #

你的代码无法运行主要有两个原因：

1. 路由函数中未定义app变量，Sanic中需通过request.app获取当前应用实例
2. app.add_task()需要传入协程对象，而非函数本身，直接传入task_to_run_in_bg会导致Sanic无法执行该任务

快速修复：路由内直接调度任务 #

修正后的路由代码如下：

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):
    payload = request.args
    size = int(payload.get('size', 10))
    start = int(payload.get('start', 0))
    response = await AdminInfoManager.get_all_applications(size, start)
    # 修正：通过request.app获取实例，传入协程对象
    request.app.add_task(task_to_run_in_bg(request.app))
    return send_response(response)

推荐方案：在管理器类中实现任务调度 #

要在管理器类中直接访问app实例并调度任务，有两种常用方式：

方式1：初始化管理器时传入app实例 #

class AdminInfoManager:
    def __init__(self, app):
        self.app = app  # 直接持有app实例

    async def get_all_applications(self, size, start):
        # 你的业务逻辑实现
        return {"data": [], "total": 0}

    async def run_background_task(self):
        # 后台任务逻辑，直接使用self.app访问应用实例
        await asyncio.sleep(2)
        logger.info('Starting background task from manager')
        for item in ['one', 'two', 'three']:
            logger.info(f'Cycle {item}')
            await asyncio.sleep(2)
        logger.info('Background task finished')

# 应用初始化时创建管理器实例并传入app
app = Sanic("my_app")
admin_manager = AdminInfoManager(app)
admin = Blueprint("admin", version=6)

@admin.route("admin/applications", methods=["GET"], name="get_apps")
async def get_all_apps(request: Request):
    payload = request.args
    size = int(payload.get('size', 10))
    start = int(payload.get('start', 0))
    response = await admin_manager.get_all_applications(size, start)
    # 通过管理器调度后台任务
    request.app.add_task(admin_manager.run_background_task())
    return send_response(response)

方式2：利用Sanic的app.ctx存储管理器（更灵活） #

适合需要延迟初始化或多场景复用的情况：

class AdminInfoManager:
    def __init__(self):
        self.app = None

    def init_app(self, app):
        self.app = app
        app.ctx.admin_manager = self  # 将管理器绑定到应用上下文

    async def get_all_applications(self, size, start):
        # 你的业务逻辑实现
        return {"data": [], "total": 0}

    async def run_background_task(self):
        # 直接使用self.app访问应用实例
        await asyncio.sleep(2)
        logger.info('Starting background task from manager')
        for item in ['one', 'two', 'three']:
            logger.info(f'Cycle {item}')
            await asyncio.sleep(2)
        logger.info('Background task finished')

# 初始化流程
app = Sanic("my_app")
admin_manager = AdminInfoManager()
admin_manager.init_app(app)  # 绑定到app上下文

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

@admin.route("admin/applications", methods=["GET"], name="get_apps")
async def get_all_apps(request: Request):
    payload = request.args
    size = int(payload.get('size', 10))
    start = int(payload.get('start', 0))
    # 从应用上下文获取管理器
    manager = request.app.ctx.admin_manager
    response = await manager.get_all_applications(size, start)
    # 调度后台任务
    request.app.add_task(manager.run_background_task())
    return send_response(response)

关键注意事项 #

• 后台任务会在应用运行期间持续执行，需添加异常捕获逻辑避免任务崩溃
• add_task()返回的asyncio.Task对象可用于跟踪任务状态或主动取消任务
• 若任务无需访问app实例，可简化为无参数协程，直接传入request.app.add_task(task_func())

内容的提问来源于stack exchange，提问作者era5tone