Python | Web フレームワーク：背景タスク

概要（背景タスク＝「レスポンスの後で、裏でこっそり動く仕事」）

背景タスクは、

「HTTP レスポンスはすぐ返したいけど、
そのあとにやりたい処理がまだ残っている」

ときに使う仕組みです。

例えば、

ログを外部サービスに送る
完了メールを送信する
重い集計処理を“結果だけ返してあとでゆっくり”やる

といった、「ユーザーにはすぐ OK を返したいが、裏ではまだ仕事がある」ケースで役立ちます。

FastAPI には BackgroundTasks という簡易な仕組みが用意されていて、
「レスポンスを返した後に実行する関数」を登録できます。

ここから、概念 → 基本コード → 実用パターン → 限界と注意点
の順でかみ砕いていきます。

なぜ背景タスクが必要になるのか（同期処理だけだと起きる問題）

全部レスポンスの前にやると「待ち時間地獄」になる

まず、背景タスクを使わない“素直な”コードを考えます。

例えば、「ユーザー登録 → メール送信」の処理。

from fastapi import FastAPI

app = FastAPI()

def send_welcome_email(email: str):
    # 実際には SMTP や外部APIでメール送信
    import time
    time.sleep(5)  # メール送信に時間がかかる想定

@app.post("/signup")
def signup(email: str):
    # ユーザーをDBに保存したり、いろいろやる
    send_welcome_email(email)
    return {"message": "signed up"}

from fastapi import FastAPI

app = FastAPI()

def send_welcome_email(email: str):
    # 実際には SMTP や外部APIでメール送信
    import time
    time.sleep(5)  # メール送信に時間がかかる想定

@app.post("/signup")
def signup(email: str):
    # ユーザーをDBに保存したり、いろいろやる
    send_welcome_email(email)
    return {"message": "signed up"}

このコードだと、

ユーザー「/signup を叩く」
サーバー「ユーザー保存 → メール送信（5秒待つ） → レスポンス」

になるので、ユーザーは 5 秒以上待たされます。

「メール送信が終わるまでレスポンスを返さない」
というのは、体感としてかなりストレスです。

重要なのは、「ユーザー視点では、“ユーザー登録が成功した”と分かればいい」ことです。
メール送信がいつ終わるかまでは、あまり気にしません。

そこで出てくるのが背景タスクです。

FastAPI の BackgroundTasks の基本構造

関数と引数を「後で実行してね」と登録するだけ

FastAPI には BackgroundTasks という仕組みがあり、
エンドポイントの引数に受け取ることで使えます。

先ほどのサインアップを、背景タスクで書き直してみます。

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_welcome_email(email: str):
    import time
    time.sleep(5)
    print(f"Welcome email sent to {email}")

@app.post("/signup")
def signup(email: str, background_tasks: BackgroundTasks):
    # ここで DB にユーザーを保存する、などの処理をする
    background_tasks.add_task(send_welcome_email, email)
    return {"message": "signed up"}

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_welcome_email(email: str):
    import time
    time.sleep(5)
    print(f"Welcome email sent to {email}")

@app.post("/signup")
def signup(email: str, background_tasks: BackgroundTasks):
    # ここで DB にユーザーを保存する、などの処理をする
    background_tasks.add_task(send_welcome_email, email)
    return {"message": "signed up"}

このようにすると、

サーバー「ユーザー保存 → 背景タスク登録 → すぐレスポンス返す」
レスポンス返却後「裏で send_welcome_email(email) が実行される」

という動きになります。

ポイントは、

背景タスクの関数（send_welcome_email）は普通の関数
background_tasks.add_task(関数, 引数...) で「後で実行してね」と登録する
レスポンスはタスク実行を待たずに返される

という流れです。

ここで、「関数そのものを登録する」という感覚を掴んでおいてください。
「今すぐ呼び出す」のではなく、「後で呼び出してもらうように登録する」というイメージです。

もう少し実務寄りの例（ログ・通知・外部連携を背景タスクに逃がす）

例１：重いログ送信を背景に回す

例えば、「エンドポイントの処理結果を外部のログサービスに送る」処理があり、
それが微妙に重いとします。

def send_log_to_external_service(user_id: int, action: str):
    import time
    time.sleep(2)
    print(f"Logged: user={user_id}, action={action}")

def send_log_to_external_service(user_id: int, action: str):
    import time
    time.sleep(2)
    print(f"Logged: user={user_id}, action={action}")

これを毎回同期的にやると、
ユーザーは余計な待ち時間を払うことになります。

背景タスクにするとこうなります。

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post("/purchase")
def purchase_item(user_id: int, item_id: int, background_tasks: BackgroundTasks):
    # ここで購入処理（DB 更新など）を行う
    background_tasks.add_task(send_log_to_external_service, user_id, f"purchase:{item_id}")
    return {"message": "purchased", "item_id": item_id}

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post("/purchase")
def purchase_item(user_id: int, item_id: int, background_tasks: BackgroundTasks):
    # ここで購入処理（DB 更新など）を行う
    background_tasks.add_task(send_log_to_external_service, user_id, f"purchase:{item_id}")
    return {"message": "purchased", "item_id": item_id}

ユーザー視点では、

購入処理が終わった時点でレスポンス
ログ送信はその後に裏で行われる

という動きになります。

重要なのは、「ユーザーに必要な処理」と「後からでいい処理」を分けることです。
背景タスクに向いているのは、「結果に影響しない、後始末や付随的な処理」です。

例２：完了通知（メール・Slack）なども背景タスク向き

何か重い処理が完了したことを、
Slack やメールで通知したいことがあります。

処理本体は同期的に動かし、
通知だけ背景タスクに逃がすパターンも多いです。

def notify_slack(message: str):
    # Slack API を叩く処理
    ...

@app.post("/job")
def run_job(background_tasks: BackgroundTasks):
    # 重い処理
    result = "something done"
    background_tasks.add_task(notify_slack, f"Job finished: {result}")
    return {"status": "accepted"}

def notify_slack(message: str):
    # Slack API を叩く処理
    ...

@app.post("/job")
def run_job(background_tasks: BackgroundTasks):
    # 重い処理
    result = "something done"
    background_tasks.add_task(notify_slack, f"Job finished: {result}")
    return {"status": "accepted"}

「ユーザーに OK と返すタイミング」と
「通知を送るタイミング」を切り分けることで、
レスポンスの体感速度を改善できます。

BackgroundTasks の「できること」と「限界」をちゃんと理解する

できること：レスポンス後に同じプロセス内で処理を続ける

BackgroundTasks は、非常にシンプルな仕組みです。

レスポンスを返した後に、
同じ Uvicorn / Gunicorn プロセス内で関数を実行する

これだけです。

つまり、

あくまで「同じアプリプロセスの中で、後回しにする」だけで、
別プロセスや別マシンに送るわけではない

ということです。

このため、

ちょっと重い I/O（メール、API 呼び出し、ログ送信など）
ユーザー体験に直結しない処理

には向いています。

限界１：サーバープロセスが落ちるとタスクも消える

BackgroundTasks は「アプリの中で」動いているので、

サーバーが再起動した
プロセスがクラッシュした

といった場合、その時点で走っていた背景タスクは消えます。

キューに溜めて再実行するような仕組みはありません。

「絶対に失敗させたくない重いバッチ」や
「時間を指定して実行するタスク」などには、

Celery
RQ
dramatiq

のようなジョブキュー／ワーカーの仕組みを使う方が安全です。

BackgroundTasks は、

軽めで、多少失敗しても致命的でない処理を
「とりあえず後ろで流す」ための簡便な手段

ととらえるのがちょうどいいです。

限界２：長時間かかるタスクには不向き

BackgroundTasks に、
数分〜数十分かかるような処理を入れるのはあまりおすすめしません。

ワーカー数の上限やタイムアウトに引っかかったり、
プロセスのライフサイクルに巻き込まれたりします。

イメージとしては、

数百ミリ秒〜数秒くらい
長くても十数秒以内

に終わる処理を想定しておくのが無難です。

それ以上の「ガチ重いバッチ」は、
別途バッチ基盤やジョブキューで扱うほうが、
設計として健全です。

背景タスクと asyncio（非同期）との関係を軽く整理しておく

sync 関数も async 関数も両方登録できる

BackgroundTasks に登録する関数は、
基本は普通の sync 関数で問題ありません。

内部で requests や smtplib を使っても構いません。

もしアプリ全体を非同期化していて、

非同期のメール送信関数
非同期の外部 API クライアント

などを使いたい場合は、
async 関数を登録することもできます（FastAPI が面倒を見てくれます）。

ただし、初心者の段階では、

背景タスクにはまず「普通の関数」を渡す
async を使うのは、その後でいい

くらいの感覚で十分です。

重要なのは、「いつ実行されるのか」の理解であって、
最初から非同期構文を複雑にしないことです。

まとめ（背景タスクは「応答を早くするための後片付けエリア」）

FastAPI の背景タスク（BackgroundTasks）を整理すると、こうなります。

• 背景タスクは、「HTTP レスポンスを返したあとに、同じプロセス内で実行される仕事」を登録する仕組みで、ユーザーの待ち時間を減らすための道具。
• background_tasks.add_task(func, *args, **kwargs) と書くだけで、「この関数をあとで実行してね」と登録できる。登録されたタスクはレスポンス後に裏で動く。
• メール送信・ログ送信・通知・外部サービス連携など、「結果には影響しないがやっておきたい処理」を背景タスクに逃がすと、レスポンスが体感的に速くなる。
• BackgroundTasks はあくまで「同じアプリプロセス内での後回し」であり、プロセスが落ちるとタスクも消える。長時間・重要バッチにはキュー系（Celery 等）のほうが向いている。
• 「今すぐユーザーに返すべき仕事」と「後でよい仕事」を切り分け、その「後でよい仕事」を BackgroundTasks に追い出す感覚を身につけると、Web API の体験が一段レベルアップする。