前言

咱们知道界说后台使命，咱们只需要在路由函数中增加详细的后台使命即可，如下代码所示：

from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def back_tasks_notification(message=""):
    pass
@app.post("/send-backtask/")
async def backtask(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(back_tasks_notification, message=f"后台使命...{message}")
    return {"message": "BackgroundTasks履行"}

其实上面的代码履行也是非常高雅可观。不过关于一些使命界说来说，可能就相对的分散杂乱，没有进行统一管理，所以下面我个人习惯把一些或一组的后台使命以类的方式来界说。

实现

首要咱们界说用于增加后台使命类，如下代码所示：

import functools
import typing
from typing import Optional, List
from fastapi import BackgroundTasks
class BackgroundTasksDependencyInject:
    def __init__(self, only_run_first: bool = True) -> None:
        # 访问路由组时候是否仅仅履行一次后台使命
        self.only_run_first: bool = only_run_first
        # 后台使命
        self.background_task = None
        # 寄存使命列表
        self.tasks: Optional[List[typing.Callable]] = []
    def add_task(self, function=None, *args, **kwargs) -> None:
        if function:
            func_obj = functools.partial(function, *args, **kwargs)
            self.tasks.append(func_obj)
    def on_task(self, *args, **kwargs) -> typing.Callable:  # pragma: nocover
        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_task(func, *args, **kwargs)
            return func
        return decorator
    async def __call__(self, background_task: BackgroundTasks) -> Optional[BackgroundTasks]:
        """
        """
        pass
        if not self.background_task:
            self.background_task = background_task
        if self.only_run_first:
            tasks = [self.background_task.add_task(func_obj) for func_obj in self.tasks]
        else:
            tasks = [background_task.add_task(func_obj) for func_obj in self.tasks]
        # print("tasks_len", len(tasks))
        return background_task

在上面的代码中，首要界说了相关的特点，它们主要包含了：

• only_run_first 判断这些使命组是否仅仅履行一次的符号
• self.tasks 后台使命的列表

其间类的包含的办法以及内部装饰器有：

• add_task 表明增加后台使命
• on_task 界说装饰器，在内部其实是对add_task的调用
• call 则是要害中的要害，它能够把类的实例目标转化为可调用目标，在内进行使命的调用的循环增加到BackgroundTasks上。

用法

在上面界说完结咱们的后台使命管理类后，接下来则是创建详细的使命组目标并，增加详细的使命，如下代码所示：

文本代码如下：

import asyncio
import time
from snowy_src.snowy_common.snowy_dependencies.inject_backgrond_tasks import BackgroundTasksDependencyInject
newtasks = BackgroundTasksDependencyInject(only_run_first=False)
# 后台使命注册
@newtasks.on_task()
async def _background_task_1():
    print("background_task01")
    await asyncio.sleep(1)
    pass
# 后台使命注册
@newtasks.on_task()
def _background_task_2():
    print("background_task02")
    time.sleep(2)
    pass

在前面咱们的知道，此刻咱们的newtask目标是一个可调用目标，能够对它进行（）调用，所以它也能够是一个依靠项，所以，咱们能够把它作为一个依靠注入项，进行运用，如下代码图所示：

需注意点：相关后台使命需要是正常呼应逻辑处理完结才会履行。

终究咱们能够在控制台看到咱们的使命被履行，如下图所示：

至此，关于【-以类的方式界说后台使命依靠项】相关介绍共享已完结！以上内容共享纯属个人经验，仅供参考！

文笔有限，如有笔误或错误！欢迎批评指正！感谢各位大佬！有什么问题也能够随时沟通！

结尾

END

简书：www.jianshu.com/u/d6960089b…

：/user/296393…

大众号：微信搜【程序员小钟同学】

新开QQ群号，欢迎随时加群沟通，相互学习。QQ群号：247491107

小钟同学 | 文 【欢迎一同学习沟通】| QQ：308711822