IT & ライフハックブログ｜学びと実践のためのアイデア集

毎日に役立つライフハックをサクッと紹介 Support by AI

システム開発

FastAPIで実践する通知・メール送信基盤設計入門：BackgroundTasks・ジョブキュー・テンプレート・再送・監査まで整える実務パターン

投稿者 greeden #FastAPI, #Python, #システム開発
green snake
Photo by Pixabay on Pexels.com
目次

FastAPIで実践する通知・メール送信基盤設計入門：BackgroundTasks・ジョブキュー・テンプレート・再送・監査まで整える実務パターン

要約

  • 通知・メール送信は、ユーザー登録や問い合わせ返信、請求、SLAアラートなど多くの業務に関わる重要な基盤です。FastAPIでは、軽い後処理なら BackgroundTasks、信頼性や再試行が必要な処理ならCeleryなどのジョブキューを使い分けると安定します。FastAPI公式では、BackgroundTasks はレスポンス後に実行したい処理に使えると説明されています。
  • メール送信は、SMTPで直接送る方法と、外部メール配信APIを使う方法があります。Pythonの非同期SMTPクライアントである aiosmtplib は、send() コルーチンをメール送信の主要な入口として案内しています。
  • 実務では、メール本文をその場で組み立てて即送信するだけでは足りません。テンプレート、送信履歴、再送、重複防止、配信失敗、購読解除、監査ログ、メトリクスまでを含めて「通知基盤」として設計することが大切です。
  • 重要な通知は、APIレスポンス中に送信完了まで待たず、まず通知ジョブを作成し、後続処理として送る設計が安全です。Celeryは、リアルタイム処理とスケジューリングを支える分散タスクキューとして公式に説明されています。
  • 本記事では、FastAPIで通知・メール送信基盤を作るときの考え方を、用途整理、テンプレート設計、同期/非同期の使い分け、Outboxパターン、再送、監査、テストまで順番に整理します。

誰が読んで得をするか

個人開発・学習者さん

ユーザー登録後の確認メール、問い合わせ受付メール、パスワードリセットメールなどをFastAPIで送りたい方に向いています。最初は send_email() を直接呼ぶだけでも動きますが、少し運用が始まると「送れなかったときどうするか」「同じメールを2回送らないか」「テンプレートをどう管理するか」が課題になります。

この記事では、まず BackgroundTasks から始め、必要に応じてジョブキューへ進む段階的な設計を紹介します。FastAPIの依存関係システムを使うと、メール送信クライアントや設定を共通化しやすくなります。

小規模チームのバックエンドエンジニアさん

問い合わせ返信、請求通知、管理画面からの一括メール、社内通知などが増え、メール送信処理がルーターやサービス層に散らばってきたチームに向いています。

通知基盤を共通化すると、テンプレート、送信履歴、再送、監査ログ、配信失敗時の対応が一箇所にまとまります。チームで「重要通知は必ず履歴に残す」「軽い通知だけBackgroundTasks」「大量送信はジョブキュー」といったルールを共有しやすくなります。

SaaS開発チーム・スタートアップの皆さま

顧客向けメール、プロダクト内通知、Webhook、Slack通知、請求通知、SLAアラートなど、複数種類の通知を扱うチームに向いています。

この段階では、メール送信は単なる補助機能ではなく、顧客体験と運用の中心です。通知の遅延、重複送信、失敗時の再送、配信停止、監査証跡まで含めて設計する必要があります。Celeryのような分散タスクキューは、大量メッセージ処理やスケジューリングを含む通知基盤と相性が良いです。

アクセシビリティ評価

  • 冒頭に要約を置き、通知基盤の全体像を先に把握できる構成にしています。
  • 「メール」「通知」「ジョブ」「テンプレート」「Outbox」などの用語は、初出時に短く説明しています。
  • コード例は短めに分け、1つのブロックで1つの役割だけを示しています。
  • 個人開発、小規模チーム、SaaSチームそれぞれの導入段階を分けているため、自分の状況に近い部分から読めます。
  • 目標レベルはAA相当です。

1. 通知基盤は「メールを送る関数」ではない

FastAPIでメール送信を実装しようとすると、最初は次のような関数を書きたくなります。

def send_email(to: str, subject: str, body: str) -> None:
    ...

もちろん、最初の一歩としてはこれで十分です。
ただし、実務ではすぐに次のような課題が出てきます。

  • メール本文をどこで管理するか
  • 同じ通知を2回送らないか
  • 送信に失敗したとき再送するか
  • ユーザーが配信停止している場合はどうするか
  • 請求通知やパスワードリセットなど重要度の違いをどう扱うか
  • 送信履歴をどう残すか
  • 顧客から「届いていない」と言われたときに追跡できるか

つまり、メール送信は「関数」ではなく「基盤」として設計した方が、後から楽になります。

2. 通知の種類を最初に分類する

通知基盤を作る前に、まず通知の種類を分類します。
すべてを同じ扱いにすると、重要通知と軽い通知が混ざってしまいます。

2.1 トランザクションメール

ユーザー操作や業務イベントに直接ひもづくメールです。

例:

  • メールアドレス確認
  • パスワードリセット
  • 問い合わせ受付
  • CS返信
  • 請求書発行
  • 支払い失敗通知
  • 契約変更通知

これは重要度が高く、送信履歴や再送が必要になりやすいです。

2.2 マーケティングメール

キャンペーンやお知らせ、ニュースレターなどです。

例:

  • 新機能紹介
  • キャンペーン案内
  • 月次ニュースレター

配信停止やセグメント管理、送信数制御が重要です。

2.3 社内通知

社内メンバーや運営チーム向けの通知です。

例:

  • 新規問い合わせ通知
  • urgentチケット通知
  • 請求失敗アラート
  • 外部API障害通知
  • SLA違反通知

メールだけでなく、SlackやWebhookに流すこともあります。

3. FastAPIでの基本方針：送信処理をルーターに直書きしない

ルーターの中で直接SMTPや外部APIを呼ぶと、だんだん見通しが悪くなります。

避けたい形は次のようなものです。

@router.post("/signup")
def signup(payload: SignupRequest):
    # ユーザー作成
    # その場でメール本文を作る
    # その場でSMTP送信
    return {"ok": True}

おすすめは、次のような責務分離です。

  • ルーター
    • リクエストを受ける
  • サービス層
    • ユーザー作成や問い合わせ返信などの業務処理をする
  • 通知サービス
    • 通知イベントを作る
  • メールクライアント
    • 実際にSMTPや外部APIへ送る
  • ジョブキュー
    • 重い送信や再送を担う

FastAPIの依存関係システムは、このような部品の注入に向いています。公式ドキュメントでも、依存関係は外部コンポーネントを統合しやすくする仕組みとして説明されています。

4. 最小のメール送信クライアントを作る

まずは、メール送信を専用クラスに閉じ込めます。
ここでは、SMTPで送る想定のインターフェースだけを示します。

from dataclasses import dataclass

@dataclass
class EmailMessage:
    to: str
    subject: str
    text_body: str
    html_body: str | None = None

class EmailClient:
    async def send(self, message: EmailMessage) -> None:
        ...

実装を外に隠しておくことで、あとからSMTPから外部メール配信APIへ切り替えやすくなります。

5. aiosmtplibで非同期SMTP送信を行う例

FastAPIの非同期エンドポイントと組み合わせるなら、非同期SMTPクライアントを使う選択肢があります。aiosmtplib のドキュメントでは、send() コルーチンがメール送信の主要な入口として推奨されています。

from email.message import EmailMessage as PyEmailMessage
import aiosmtplib

class SMTPEmailClient:
    def __init__(
        self,
        hostname: str,
        port: int,
        username: str | None = None,
        password: str | None = None,
        use_tls: bool = True,
    ):
        self.hostname = hostname
        self.port = port
        self.username = username
        self.password = password
        self.use_tls = use_tls

    async def send(self, message: EmailMessage) -> None:
        email = PyEmailMessage()
        email["From"] = "no-reply@example.com"
        email["To"] = message.to
        email["Subject"] = message.subject
        email.set_content(message.text_body)

        if message.html_body:
            email.add_alternative(message.html_body, subtype="html")

        await aiosmtplib.send(
            email,
            hostname=self.hostname,
            port=self.port,
            username=self.username,
            password=self.password,
            use_tls=self.use_tls,
        )

この実装をルーターに直接書かず、通知サービスから呼び出すようにすると保守しやすくなります。

6. テンプレート設計：本文をコードに埋め込まない

メール本文をPythonコードに直接書くと、文面修正のたびにコードを触る必要があります。
通知が増えるほど、テンプレートを分けた方が扱いやすくなります。

6.1 テンプレートキーを決める

例:

  • user.email_verification
  • user.password_reset
  • ticket.created
  • ticket.replied
  • billing.payment_failed
  • subscription.canceled

このように、<domain>.<event> の形にすると整理しやすいです。

6.2 テンプレート入力モデルを決める

from pydantic import BaseModel, EmailStr

class TicketReplyEmailContext(BaseModel):
    customer_email: EmailStr
    ticket_id: int
    subject: str
    reply_body: str

テンプレートに渡す値をモデル化すると、必要な変数が明確になります。
「テンプレート側では reply_body が必要なのに、呼び出し側で渡し忘れた」という事故を減らせます。

7. 通知サービスを作る

メールクライアントを直接呼ぶのではなく、通知サービスを挟みます。

class NotificationService:
    def __init__(self, email_client: EmailClient):
        self.email_client = email_client

    async def send_ticket_reply_email(self, context: TicketReplyEmailContext) -> None:
        message = EmailMessage(
            to=context.customer_email,
            subject=f"Re: {context.subject}",
            text_body=context.reply_body,
            html_body=f"<p>{context.reply_body}</p>",
        )
        await self.email_client.send(message)

この形にしておくと、

  • テンプレート選択
  • 配信停止確認
  • 送信履歴保存
  • メトリクス記録
  • 監査ログ

を通知サービスにまとめやすくなります。

8. FastAPIの依存関数で通知サービスを注入する

FastAPIでは、依存関数を使って通知サービスを組み立てると便利です。

from fastapi import Depends

def get_email_client() -> EmailClient:
    return SMTPEmailClient(
        hostname="smtp.example.com",
        port=465,
        username="smtp-user",
        password="smtp-password",
    )

def get_notification_service(
    email_client: EmailClient = Depends(get_email_client),
) -> NotificationService:
    return NotificationService(email_client=email_client)

ルーター側では次のように使えます。

from fastapi import APIRouter, Depends

router = APIRouter(prefix="/admin/tickets", tags=["admin-tickets"])

@router.post("/{ticket_id}/reply")
async def reply_ticket(
    ticket_id: int,
    notification_service: NotificationService = Depends(get_notification_service),
):
    context = TicketReplyEmailContext(
        customer_email="customer@example.com",
        ticket_id=ticket_id,
        subject="お問い合わせについて",
        reply_body="ご連絡ありがとうございます。",
    )
    await notification_service.send_ticket_reply_email(context)
    return {"status": "sent"}

ただし、このままだとレスポンス中に送信完了まで待ちます。
軽い通知でも、ユーザー体験を考えると後続処理へ逃がす方が良いことが多いです。

9. BackgroundTasksで軽い通知をレスポンス後に回す

FastAPI公式では、BackgroundTasks を使うと、レスポンス後に実行したい処理を定義できると説明されています。メール通知やログ送信のように、クライアントが完了を待たなくてよい処理に向いています。

from fastapi import BackgroundTasks

def send_ticket_reply_email_sync(ticket_id: int, email: str) -> None:
    # 実際にはメール送信処理
    pass

@router.post("/{ticket_id}/reply")
def reply_ticket(
    ticket_id: int,
    background_tasks: BackgroundTasks,
):
    # 返信メッセージをDBへ保存する処理

    background_tasks.add_task(
        send_ticket_reply_email_sync,
        ticket_id,
        "customer@example.com",
    )

    return {"status": "reply_saved"}

BackgroundTasks はとても便利ですが、信頼性が必要な通知には注意が必要です。
同一プロセス内で動くため、プロセス停止や大量送信には弱いです。重要通知や再送が必要な通知は、ジョブキューへ移すことを検討します。

10. 重要通知はOutboxパターンで扱う

Outboxパターンとは、送信すべき通知をまずDBに保存し、その後でワーカーが送信する設計です。
これにより、業務処理とメール送信の整合性を保ちやすくなります。

10.1 Outboxテーブルのイメージ

from datetime import datetime
from pydantic import BaseModel
from typing import Literal

NotificationStatus = Literal["pending", "sending", "sent", "failed"]

class NotificationOutboxItem(BaseModel):
    id: int
    type: str
    recipient: str
    payload: dict
    status: NotificationStatus
    retry_count: int = 0
    created_at: datetime
    sent_at: datetime | None = None

10.2 業務処理では「送る」ではなく「積む」

def enqueue_ticket_reply_notification(ticket_id: int, customer_email: str, reply_body: str) -> None:
    # 実際はDBへINSERT
    outbox_item = {
        "type": "ticket.replied",
        "recipient": customer_email,
        "payload": {
            "ticket_id": ticket_id,
            "reply_body": reply_body,
        },
        "status": "pending",
    }

この設計にすると、返信保存は成功したのにメール送信だけ失敗した場合でも、後からOutboxを見て再送できます。

11. Celeryなどのジョブキューへ送る

通知が多い、再試行したい、失敗時に追跡したい場合は、ジョブキューが向いています。Celeryは、メッセージ処理のための分散システムであり、リアルタイム処理とスケジューリングに焦点を当てたタスクキューとして公式に説明されています。

11.1 ジョブ化する対象

  • 請求失敗通知
  • パスワードリセットメール
  • CSV生成完了通知
  • 一括メール送信
  • SLA違反通知
  • CS返信通知
  • リマインド通知

11.2 Celeryタスクの例

from celery import Celery

celery_app = Celery(
    "notifications",
    broker="redis://redis:6379/0",
    backend="redis://redis:6379/1",
)

@celery_app.task(
    name="notifications.send_email",
    autoretry_for=(Exception,),
    retry_kwargs={"max_retries": 3},
)
def send_email_task(outbox_id: int) -> None:
    # outbox_id からDBを引き、メール送信
    # 成功したらstatus=sent
    # 失敗したらCeleryのretryへ
    pass

このようにOutboxとCeleryを組み合わせると、送信履歴と再試行の両方を扱いやすくなります。

12. 重複送信を防ぐ：冪等性キーを持つ

通知で怖いのは、同じメールを何度も送ってしまうことです。
とくに請求やパスワードリセット、契約変更通知では、重複送信が顧客不安につながります。

そこで、通知には冪等性キーを持たせます。

例:

ticket_reply:{ticket_id}:{message_id}
password_reset:{user_id}:{token_id}
payment_failed:{invoice_id}

Outboxに dedup_key を持たせ、一意制約を付けると安全です。

class NotificationOutboxItem(BaseModel):
    id: int
    dedup_key: str
    type: str
    recipient: str
    payload: dict
    status: NotificationStatus

同じ dedup_key の通知が既にあれば、新規作成しないようにします。
これにより、リトライや再送処理が入っても、二重登録を防ぎやすくなります。

13. 配信停止・通知設定を必ず考える

通知基盤では、「送る」だけでなく「送らない」設計も重要です。

ユーザーやテナントごとに、次のような設定を持つことがあります。

  • プロダクト通知を受け取る
  • 請求通知を受け取る
  • マーケティングメールを受け取る
  • SLAアラートを受け取る
  • 週次サマリーを受け取る

ただし、すべての通知を停止できるわけではありません。
たとえば、パスワードリセットや請求失敗など、アカウント維持に必要なトランザクションメールは、マーケティングメールとは別扱いにする必要があります。

13.1 通知カテゴリを分ける

NotificationCategory = Literal[
    "transactional",
    "billing",
    "security",
    "marketing",
    "internal",
]

配信停止の扱いはカテゴリごとに変えます。

def can_send_notification(category: str, user_settings: dict) -> bool:
    if category in {"transactional", "security"}:
        return True
    if category == "marketing":
        return bool(user_settings.get("marketing_opt_in"))
    return True

このように分類しておくと、運用ルールを実装へ落とし込みやすくなります。

14. 送信履歴と監査ログを分ける

通知基盤では、送信履歴と監査ログを混同しない方がよいです。

送信履歴

  • どの通知を送ろうとしたか
  • 誰に送ったか
  • 成功したか失敗したか
  • 何回リトライしたか
  • 外部メールサービスのメッセージID

監査ログ

  • 誰の操作によって通知が発生したか
  • どの業務イベントに基づく通知か
  • 重要操作として残すべきか

たとえば、CS返信メールなら、

  • 送信履歴: 顧客へメール送信した履歴
  • 監査ログ: CS担当者がチケットへ返信した履歴

という違いがあります。
両方を残すと、問い合わせ対応や障害調査がかなり楽になります。

15. メール本文に個人情報や機密情報を入れすぎない

メールは一度送ると取り戻せません。
また、転送やスクリーンショットも簡単にできます。

そのため、メール本文には機密情報を詰め込みすぎない方が安全です。

避けたいもの:

  • パスワードそのもの
  • 長期間有効なトークン
  • 詳細な請求情報全文
  • 個人情報の過剰な記載
  • 内部メモ

パスワードリセットメールでは、短時間だけ有効なURLを送り、詳細な処理はログイン後の画面で行う方が安全です。

16. 通知テンプレートのバージョン管理

テンプレートは、時間とともに変わります。

  • 件名を変える
  • 文面を改善する
  • フッターを追加する
  • 法務文言を入れる
  • 多言語対応する

そのため、重要な通知では「どのテンプレートバージョンで送ったか」を履歴に残すと便利です。

class NotificationOutboxItem(BaseModel):
    id: int
    type: str
    template_key: str
    template_version: str
    recipient: str
    payload: dict
    status: NotificationStatus

これがあると、後から「この時期のメールにはどの文面が入っていたか」を追いやすくなります。

17. 多言語対応を見据えたテンプレート設計

SaaSでは、将来的に日本語・英語など複数言語の通知が必要になることがあります。
その場合、テンプレートキーに言語を混ぜるより、テンプレートキーとロケールを分ける方が整理しやすいです。

template_key = "ticket.replied"
locale = "ja"

実装側では、

def resolve_template(template_key: str, locale: str) -> str:
    return f"templates/{locale}/{template_key}.html"

のように組み立てられます。

18. 通知のメトリクスを取る

通知は「送れたかどうか」が非常に重要です。
最低限、次のメトリクスを見られるようにしておくと安心です。

  • 通知作成数
  • 送信成功数
  • 送信失敗数
  • 再試行回数
  • Outbox滞留件数
  • 平均送信遅延
  • カテゴリ別送信数

たとえば、Outboxに created_at と sent_at を持っておくと、送信遅延を計算できます。

delay_seconds = (sent_at - created_at).total_seconds()

通知はユーザー体験に直結します。
「送れていない」ことを早く検知できるようにしましょう。

19. テスト方針：実際にメールを送らずに検証する

テストでは、本物のSMTPや外部メールAPIを叩かない方が安全です。

19.1 フェイクEmailClientを使う

class FakeEmailClient:
    def __init__(self):
        self.sent: list[EmailMessage] = []

    async def send(self, message: EmailMessage) -> None:
        self.sent.append(message)

これを依存関数で差し替えれば、実際に送信せずに「送ろうとした内容」を確認できます。

async def test_send_ticket_reply_email():
    fake = FakeEmailClient()
    service = NotificationService(email_client=fake)

    context = TicketReplyEmailContext(
        customer_email="customer@example.com",
        ticket_id=1,
        subject="お問い合わせ",
        reply_body="返信本文",
    )

    await service.send_ticket_reply_email(context)

    assert len(fake.sent) == 1
    assert fake.sent[0].to == "customer@example.com"

19.2 Outboxのテスト

Outboxを使う場合は、次のようなテストが重要です。

  • 通知イベントでOutboxが1件作られる
  • 同じ dedup_key では重複作成されない
  • 送信成功で status=sent になる
  • 送信失敗で retry_count が増える
  • 配信停止設定がある場合は作成されない、または skipped になる

20. よくある失敗パターン

20.1 ルーターから直接メール送信する

最初は楽ですが、テンプレート、配信停止、履歴、再送が入ると破綻しやすくなります。通知サービスへ寄せる方が安全です。

20.2 重要通知をBackgroundTasksだけで済ませる

BackgroundTasks はレスポンス後に処理できて便利ですが、プロセス停止や大量処理には弱いです。重要通知はOutboxやジョブキューを検討しましょう。

20.3 テンプレート本文をコードに埋め込む

文面修正のたびにコード変更が必要になります。テンプレートキーとコンテキストを分けるのがおすすめです。

20.4 配信停止を考えない

マーケティングメールとトランザクションメールを分けないと、法務・運用・顧客体験の面で問題になりやすいです。

20.5 送信履歴を残さない

「届いていない」と言われたときに調査できません。重要通知は、送信履歴とステータスを持ちましょう。

21. 読者別ロードマップ

個人開発・学習者さん

  1. EmailClient クラスを作る
  2. NotificationService を作る
  3. BackgroundTasks で軽い通知をレスポンス後に回す
  4. テンプレートキーを導入する
  5. フェイククライアントでテストする

小規模チームのエンジニアさん

  1. 通知種別を棚卸しする
  2. トランザクションメールとマーケティングメールを分ける
  3. Outboxテーブルを導入する
  4. 重要通知はジョブキューへ送る
  5. 送信履歴と監査ログを分けて設計する

SaaS開発チーム・スタートアップの皆さま

  1. 通知基盤を業務イベント基盤として整理する
  2. テンプレート、ロケール、バージョンを管理する
  3. 冪等性キーで重複送信を防ぐ
  4. Outbox滞留や送信失敗率を監視する
  5. CS対応、請求、SLA、Webhook連携と統合する

参考リンク

まとめ

  • 通知・メール送信は、FastAPIアプリの中で軽く見られがちですが、実務では顧客体験、請求、CS、監査、セキュリティに深く関わる重要な基盤です。
  • 最初は BackgroundTasks で軽い通知をレスポンス後に回すところから始めても大丈夫です。ただし、重要通知や再送が必要な通知はOutboxやCeleryなどのジョブキューへ進める方が安全です。
  • メール本文はテンプレート化し、通知種別、カテゴリ、テンプレートバージョン、ロケールを分けて管理すると、後からの変更に強くなります。
  • 重複送信を防ぐには冪等性キーが有効です。Outboxに一意制約を持たせると、再試行や二重イベントにも強くなります。
  • 送信履歴、監査ログ、メトリクスを整えることで、「送れたか」「なぜ送れなかったか」「誰の操作で送られたか」を追える通知基盤になります。

次の記事としては、この流れと相性が良い「FastAPIで作るSLA・エスカレーション管理API設計」や、「FastAPIで実践するイベント駆動アーキテクチャ入門」が自然につながります。

投稿者 greeden

関連投稿

システム開発

公開して終わりにしないWebアクセシビリティ運用｜WCAG 2.2時代の試験・改善・体制づくり

greeden
システム開発

Amazon Athena徹底解説：BigQuery・Azure Synapse Serverless比較でわかる“サーバーレスSQL分析基盤”の選び方

greeden
システム開発

【実務完全ガイド】Laravelで作る通知センター――Database Notifications、既読管理、メール連携、リアルタイム通知、アクセシブルな通知UI

greeden

コメントを残す

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です

日本語が含まれない投稿は無視されますのでご注意ください。（スパム対策）

見逃しています

ショッピング

梅雨・猛暑対策家電おすすめ10選｜部屋干し、空気循環、外出時の暑さまでまとめて比較

システム開発

公開して終わりにしないWebアクセシビリティ運用｜WCAG 2.2時代の試験・改善・体制づくり

ビジネス

2026年5月31日の世界主要ニュース総まとめ

ショッピング

ポータブルSSD 1TBおすすめ10選。USB-C、PS5、動画編集向けに比較