跳到内容

后台任务 - BackgroundTasks

你可以在路径操作函数或依赖项函数中声明一个类型为 BackgroundTasks 的参数，然后使用它在响应发送后调度后台任务的执行。

您可以直接从 fastapi 导入它。

from fastapi import BackgroundTasks

fastapi.BackgroundTasks 

BackgroundTasks(tasks=None)

基类： BackgroundTasks

一组将在响应发送给客户端后被调用的后台任务。

在 FastAPI 后台任务文档中了解更多信息。

示例

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}
源代码位于 starlette/background.py 
def __init__(self, tasks: Sequence[BackgroundTask] | None = None):
    self.tasks = list(tasks) if tasks else []

func 实例属性 

func = func

args 实例属性 

args = args

kwargs 实例属性 

kwargs = kwargs

is_async 实例属性 

is_async = is_async_callable(func)

tasks 实例属性 

tasks = list(tasks) if tasks else []

add_task 

add_task(func, *args, **kwargs)

添加一个在响应发送后于后台调用的函数。

在 FastAPI 后台任务文档中了解更多信息。

参数 描述
func

响应发送后要调用的函数。

它可以是常规的 def 函数或 async def 函数。

类型： Callable[P, Any]

源代码位于 fastapi/background.py 
def add_task(
    self,
    func: Annotated[
        Callable[P, Any],
        Doc(
            """
            The function to call after the response is sent.

            It can be a regular `def` function or an `async def` function.
            """
        ),
    ],
    *args: P.args,
    **kwargs: P.kwargs,
) -> None:
    """
    Add a function to be called in the background after the response is sent.

    Read more about it in the
    [FastAPI docs for Background Tasks](https://fastapi.org.cn/tutorial/background-tasks/).
    """
    return super().add_task(func, *args, **kwargs)