💳

PythonとStripeで収益化を全自動にした話

に公開
Python
個人開発
Stripe
tech

はじめに

個人開発でSaaSを作り続けて、今は16本が稼働中です。月額課金のSaaSを量産していると、毎回「Stripe連携・Webhook処理・プラン管理」を手で実装するのがしんどくなってきます。

最初の数本はコピペしながら対応していましたが、さすがに限界を感じて「収益化の部分だけ全自動にする」仕組みを作りました。今回はその設計と実装を共有します。

課題：毎回同じ実装を繰り返していた

自分が作っているSaaSはこういうものが多いです：

  • FreelancePainPollBot（¥4,350/月）：フリーランサーの困りごとを集約・可視化
  • WorkflowNagBot（¥2,850/月）：n8nやMCP連携の設定を補助
  • SoloTaxReceiptBot（¥1,350/月）：個人事業主の確定申告自動化

どれも「月額課金・ユーザー管理・機能制限」という構造は同じです。なのに毎回こういう作業をしていました：

  1. Stripeのダッシュボードで商品・プランを手動作成
  2. Webhookエンドポイントを実装
  3. サブスクリプション状態をDBに同期
  4. 未払い・解約の処理を書く

特に「Webhookを受け取ってDBを更新する」部分は、イベントの種類が多くてバグを埋め込みやすい。ある時 customer.subscription.updated を処理し忘れて、プランダウングレードが反映されないバグを本番に出してしまいました。

解決策：Stripe連携を共通ライブラリ化する

方針はシンプルです。

  • Stripe操作（顧客作成・サブスク作成・解約）を関数化
  • Webhookハンドラを一箇所にまとめてイベント網羅
  • サブスクリプション状態をシンプルなDBテーブルで管理

FastAPIとSQLiteを使って、新しいSaaSを作るときはこのボイラープレートをコピーするだけの状態にしました。

実装

Webhook処理の核心部分

一番重要なのはWebhookハンドラです。Stripeから飛んでくるイベントを漏れなく処理するのがポイント。

python
import stripe
from fastapi import FastAPI, Request, HTTPException
from sqlalchemy.orm import Session

app = FastAPI()

STRIPE_WEBHOOK_SECRET = "whsec_xxxx"

EVENT_HANDLERS = {
"customer.subscription.created": handle_subscription_created,
"customer.subscription.updated": handle_subscription_updated,
"customer.subscription.deleted": handle_subscription_deleted,
"invoice.payment_failed": handle_payment_failed,
"invoice.payment_succeeded": handle_payment_succeeded,
}

@app.post("/webhook/stripe")
async def stripe_webhook(request: Request, db: Session = Depends(get_db)):
payload = await request.body()
sig_header = request.headers.get("stripe-signature")

try:
    event = stripe.Webhook.construct_event(
        payload, sig_header, STRIPE_WEBHOOK_SECRET
    )
except stripe.error.SignatureVerificationError:
    raise HTTPException(status_code=400, detail="Invalid signature")

handler = EVENT_HANDLERS.get(event["type"])
if handler:
    handler(event["data"]["object"], db)
# 知らないイベントは無視（エラーにしない）

return {"status": "ok"}

EVENT_HANDLERS を辞書にしておくのがポイントです。if/elif で分岐すると、イベントを追加するたびに関数が肥大化します。辞書にしておけば、新しいイベントへの対応は1行追加するだけで済みます。

サブスクリプション状態の同期

python
def handle_subscription_updated(subscription: dict, db: Session):
"""
プランの変更・ステータス変化を全部ここで処理する
active / past_due / canceled を正規化して保存
"""
customer_id = subscription["customer"]
status = subscription["status"] # 'active', 'past_due', 'canceled' など
plan_id = subscription["items"]["data"][0]["price"]["id"]

record = db.query(Subscription).filter_by(stripe_customer_id=customer_id).first()
if record:
    record.status = status
    record.plan_id = plan_id
    record.current_period_end = subscription["current_period_end"]
    db.commit()

customer.subscription.updated は「プラン変更・支払い失敗後の状態変化・トライアル終了」など、様々なシーンで飛んできます。ここで status と plan_id を必ず更新するようにしておくと、他のイベントへの対応が薄くても最低限の整合性が保てます。

運用してわかったこと

よかったこと

  • 新しいSaaSの「収益化部分の実装時間」が3〜4時間から30分以下になった
  • Webhookの処理漏れによるバグがなくなった（イベントをすべて辞書で管理しているので、追加忘れに気づきやすい）
  • past_due の状態を正しくハンドルすることで、支払い失敗ユーザーへの対応が自動化できた

正直なところ

完璧ではないです。

  • Stripeのイベントは種類が多く、customer.subscription.updated の中で何が変わったのかは previous_attributes を見ないといけない。最初はここを手抜きしていてバグを出した
  • Webhookのリトライで同じイベントが2回来ることがある。べき等処理（同じイベントを2回処理しても結果が変わらない実装）を意識しないとデータが壊れる
  • SaaS本数が増えると、共通ライブラリのバージョン管理が面倒になってくる（今はコピペ運用なので、修正を各プロジェクトに手動で反映している）

まとめ

Stripe連携を「毎回書くもの」から「コピーして使うもの」に変えただけで、新しいSaaSを出すスピードが体感で2倍以上になりました。

個人開発者が複数のSaaSを運営するなら、課金周りのコードは最初に設計コストをかけて共通化しておくのが断然おすすめです。特にWebhookハンドラは「辞書でイベントをマッピングする」「べき等処理を意識する」この2点を押さえるだけで安定度が大きく上がります。

今後は各プロジェクトで手動コピーしている部分をPyPIパッケージにまとめることを検討しています。それができたらまた記事にします。

GitHubで編集を提案

Discussion