跳至內容

背景任務 - 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"}
參數 說明
tasks

類型: Sequence[BackgroundTask] | None 預設值: None

原始程式碼位於 starlette/background.py
32
33
def __init__(self, tasks: typing.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]

*args

類型: args 預設值: ()

**kwargs

類型: kwargs 預設值: {}

原始程式碼位於 fastapi/background.py
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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.dev.org.tw/tutorial/background-tasks/).
    """
    return super().add_task(func, *args, **kwargs)