👻

⚙️ FastAPI 実践入門：第九歩目で学ぶ 非同期処理とバックグラウンドタスク

に公開
Python
FastAPI
非同期処理
zennfes2025free
tech

🌐 はじめに

FastAPIはその名の通り、高速なWeb API開発を可能にするフレームワークですが、その真価は非同期処理やバックグラウンドタスクといった機能を組み合わせることで、より効率的かつスケーラブルなアプリケーションを構築できる点にあります。
この第九歩目では、クライアントからのリクエストに素早く応答しつつ、背後で画像処理やメール送信などの重い処理を行う「非同期処理」および「バックグラウンドタスク」の使い方を詳しく解説します。FastAPIを使ったアプリケーションを本番環境でも安定して動作させるために、必須の技術を身につけましょう。

🧵 非同期処理（async / await）の基本

FastAPIでは、Pythonの async def を使って非同期関数を定義できます。非同期処理の大きな利点は、重たい処理（たとえば外部APIへのリクエストやI/O処理）をブロックせずに、他の処理を並行して行えることです。
以下は3秒待機するだけの非同期処理の例です。

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/wait")
async def wait_example():
    await asyncio.sleep(3)
    return {"message": "3秒待ちました"}

この例では、処理の途中で asyncio.sleep を使って非同期的に待機しています。この間、FastAPIは他のリクエストを処理できるため、スループットが高まります。複数のクライアントが同時にアクセスするようなサービスでは、非常に重要な技術です。

🔄 バックグラウンドタスクの使い方

FastAPIでは、BackgroundTasks クラスを使って、リクエストの応答とは別に処理を行うことができます。これにより、ユーザーに対して素早く応答し、裏で時間のかかる処理を実行することが可能です。
次の例は、ログファイルに情報を書き出す処理をバックグラウンドで行うものです。

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(message + "\n")
        time.sleep(2)

@app.post("/submit/")
async def submit_data(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "ユーザーがデータを送信しました")
    return {"message": "処理を受け付けました（ログ書き込みはバックグラウンド）"}

ユーザーにはすぐにレスポンスを返し、ログファイルへの書き込みは後からバックグラウンドで行われます。

🖼️ 画像処理をバックグラウンドで行う

前回の記事で扱った画像アップロード機能と組み合わせて、アップロード後に非同期でサムネイル画像を生成する例を見てみましょう。

from fastapi import UploadFile, File
from PIL import Image
import shutil

def create_thumbnail(filename):
    image = Image.open(filename)
    image.thumbnail((128, 128))
    image.save(f"thumb_{filename}")

@app.post("/upload/")
async def upload_image(file: UploadFile = File(...), background_tasks: BackgroundTasks = None):
    file_path = f"temp_{file.filename}"
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    background_tasks.add_task(create_thumbnail, file_path)
    return {"message": "アップロード完了。サムネイル作成はバックグラウンドで行われます。"}

このように、画像処理のような重たい作業を別スレッドで処理することで、APIの応答性を保ちながら非同期に処理を進められます。

📧 メール送信やDB書き込みも非同期で

バックグラウンドタスクは、画像処理以外にもさまざまな用途に利用できます。たとえば、以下のようなケースです：

  • 📬 ユーザー登録後の確認メール送信（SendGridやSMTP）
  • 🗃️ データベースへの操作ログの記録
  • 🌐 外部APIとの連携
  • 🛎️ 通知機能の実装

以下はメール送信をバックグラウンドで行う例です。

def send_email(to: str):
    # 実際にはSMTPやSendGridのライブラリを使って実装
    print(f"{to} にメールを送信しました")

@app.post("/register")
async def register_user(background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, "test@example.com")
    return {"message": "登録完了。確認メール送信中。"}

メール送信処理をメインのフローから分離することで、ユーザー体験（UX）を損なうことなく、必要な処理を確実に実行できます。

🧩 応用：複数タスクの同時実行とエラーハンドリング

バックグラウンドタスクは1つだけでなく、複数追加することも可能です。エラーハンドリングを行うには、ログ出力やリトライ処理と組み合わせると良いでしょう。

def notify_admin():
    print("管理者に通知を送りました")

def update_statistics():
    print("統計情報を更新しました")

@app.post("/action")
async def perform_action(background_tasks: BackgroundTasks):
    background_tasks.add_task(notify_admin)
    background_tasks.add_task(update_statistics)
    return {"message": "アクションを受け付けました。バックグラウンドで処理します。"}

🎯 まとめ

  • FastAPIの async/await により、非同期処理を直感的に記述できます。
  • BackgroundTasks を用いることで、応答後に必要な処理を並列に実行でき、UXやパフォーマンスを向上させます。
  • 画像処理・メール送信・ログ記録・通知送信など、多くの処理をバックグラウンド化することで、APIのレスポンスタイムを短縮できます。
  • 本番運用では、タスクの失敗や再実行への備えも重要となるため、ログと例外処理も設計に含めましょう。

株式会社ONE WEDGE

【Serverlessで世の中をもっと楽しく】 ONE WEDGEはServerlessシステム開発を中核技術としてWeb系システム開発、AWS/GCPを利用した業務システム・サービス開発、PWAを用いたモバイル開発、Alexaスキル開発など、元気と技術力を武器にお客様に真摯に向き合う価値創造企業です。
https://onewedge.co.jp/

Discussion