FastAPI 模板

fastapi-boilerplate/
├── app/
│   ├── main.py             # FastAPI application entry point
│   ├── api/
│   │   ├── checkout.py     # Checkout session endpoint
│   │   ├── portal.py       # Customer portal endpoint
│   │   └── webhook.py      # Webhook handler endpoint
│   ├── core/
│   │   └── config.py       # Environment configuration
│   ├── lib/
│   │   ├── customers.py    # Customer payload helpers
│   │   └── products.py     # Products shown on the pricing page
│   └── templates/
│       ├── base.html       # Base template
│       └── index.html      # Pricing page
├── requirements.txt        # Python dependencies
├── .env.example            # Environment template
└── README.md

from typing import Optional

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from dodopayments import DodoPayments
from app.core.config import settings
from app.lib.customers import CustomerData, to_dodo_customer_payload

router = APIRouter()
client = DodoPayments(
    bearer_token=settings.DODO_PAYMENTS_API_KEY,
    environment=settings.DODO_PAYMENTS_ENVIRONMENT,
)

class CheckoutRequest(BaseModel):
    product_id: str
    quantity: int = 1
    customer: Optional[CustomerData] = None

@router.post("/")
async def create_checkout(request: CheckoutRequest):
    try:
        checkout_session = client.checkout_sessions.create(
            product_cart=[{
                "product_id": request.product_id,
                "quantity": request.quantity,
            }],
            customer=to_dodo_customer_payload(request.customer),
            show_saved_payment_methods=True,
            feature_flags={"allow_discount_code": True},
            return_url=settings.DODO_PAYMENTS_RETURN_URL,
        )
        return {"session_id": checkout_session.session_id, "checkout_url": checkout_session.checkout_url}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

import json

from fastapi import APIRouter, Request, HTTPException
from dodopayments import DodoPayments
from app.core.config import settings

router = APIRouter()
client = DodoPayments(
    bearer_token=settings.DODO_PAYMENTS_API_KEY,
    environment=settings.DODO_PAYMENTS_ENVIRONMENT,
    webhook_key=settings.DODO_PAYMENTS_WEBHOOK_KEY,
)

@router.post("/")
async def handle_webhook(request: Request):
    body = await request.body()

    # Dodo Payments follows the Standard Webhooks spec: the signature covers
    # `id.timestamp.body`, so all three webhook-* headers must be passed in.
    try:
        client.webhooks.unwrap(
            body,
            headers={
                "webhook-id": request.headers.get("webhook-id", ""),
                "webhook-signature": request.headers.get("webhook-signature", ""),
                "webhook-timestamp": request.headers.get("webhook-timestamp", ""),
            },
        )
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Webhook verification failed: {e}")

    event_type = json.loads(body).get("type", "")

    if event_type == "subscription.active":
        print("Subscription is active")  # Grant access, update DB, send email
    elif event_type == "payment.succeeded":
        print("Payment succeeded")  # Fulfill order, update DB, send email
    elif event_type == "subscription.cancelled":
        print("Subscription cancelled")  # Revoke access, update user status
    elif event_type == "payment.failed":
        print("Payment failed")  # Notify user, update payment status
    else:
        print(f"Unhandled event type: {event_type}")

    return {"status": "success", "event": event_type}

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from dodopayments import DodoPayments
from app.core.config import settings
from app.lib.customers import sanitize_customer_id

router = APIRouter()
client = DodoPayments(
    bearer_token=settings.DODO_PAYMENTS_API_KEY,
    environment=settings.DODO_PAYMENTS_ENVIRONMENT,
)

class PortalRequest(BaseModel):
    customer_id: str

@router.post("/")
async def create_customer_portal(request: PortalRequest):
    try:
        customer_id = sanitize_customer_id(request.customer_id)
        portal_session = client.customers.customer_portal.create(
            customer_id=customer_id,
        )
        return {"url": portal_session.link}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

ngrok http 8000

https://your-ngrok-url.ngrok.io/api/webhook/

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t fastapi-dodo .
docker run -p 8000:8000 --env-file .env fastapi-dodo