thedixitjain FastAPIパターン
本番指向のFastAPIサービスのためのパターン。
使用するタイミング
- FastAPIアプリを構築またはレビューする場合。
- ルーター、スキーマ、依存関係、データベースアクセスを分割する場合。
- データベースや外部サービスを呼び出す非同期エンドポイントを記述する場合。
- 認証、認可、OpenAPIドキュメント、テスト、またはデプロイ設定を追加する場合。
- FastAPI PRをコピー可能な例とリスクについて確認する場合。
仕組み
FastAPIアプリを明示的な依存関係とサービスコードの上の薄いHTTPレイヤーとして扱います:
main.pyはアプリ構築、ミドルウェア、例外ハンドラー、ルーター登録を担当する。schemas/はPydanticのリクエストとレスポンスモデルを担当する。dependencies.pyはデータベース、認証、ページネーション、リクエストスコープの依存関係を担当する。services/またはcrud/はビジネスと永続化操作を担当する。tests/は本番リソースを開かずに依存関係をオーバーライドする。
小さなルーターと明示的なresponse_model宣言を優先します。レスポンススキーマには生のORMオブジェクト、シークレット、フレームワークのグローバル変数を含めないでください。
プロジェクトレイアウト
app/
|-- main.py
|-- config.py
|-- dependencies.py
|-- exceptions.py
|-- api/
| `-- routes/
| |-- users.py
| `-- health.py
|-- core/
| |-- security.py
| `-- middleware.py
|-- db/
| |-- session.py
| `-- crud.py
|-- models/
|-- schemas/
`-- tests/
アプリケーションファクトリー
テストとワーカーが制御された設定でアプリをビルドできるように、ファクトリーを使用します。
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.routes import health, users
from app.config import settings
from app.db.session import close_db, init_db
from app.exceptions import register_exception_handlers
@asynccontextmanager
async def lifespan(app: FastAPI):
await init_db()
yield
await close_db()
def create_app() -> FastAPI:
app = FastAPI(
title=settings.api_title,
version=settings.api_version,
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=bool(settings.cors_origins),
allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)
register_exception_handlers(app)
app.include_router(health.router, prefix="/health", tags=["health"])
app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
return app
app = create_app()
allow_credentials=Trueと一緒にallow_origins=["*"]を使用しないでください; ブラウザはその組み合わせを拒否し、Starletteは認証情報付きリクエストに対してそれを禁止します。
Pydanticスキーマ
リクエスト、更新、レスポンスのモデルを分離します。
from datetime import datetime
from typing import Annotated
from uuid import UUID
from pydantic import BaseModel, ConfigDict, EmailStr, Field
class UserBase(BaseModel):
email: EmailStr
full_name: Annotated[str, Field(min_length=1, max_length=100)]
class UserCreate(UserBase):
password: Annotated[str, Field(min_length=12, max_length=128)]
class UserUpdate(BaseModel):
email: EmailStr | None = None
full_name: Annotated[str | None, Field(min_length=1, max_length=100)] = None
class UserResponse(UserBase):
model_config = ConfigDict(from_attributes=True)
id: UUID
created_at: datetime
updated_at: datetime
レスポンスモデルにはパスワードハッシュ、アクセストークン、リフレッシュトークン、内部認可状態を含めてはなりません。
依存関係
リクエストスコープのリソースには依存性注入を使用します。
from collections.abc import AsyncIterator
from uuid import UUID
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.security import decode_token
from app.db.session import session_factory
from app.models.user import User
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
async def get_db() -> AsyncIterator[AsyncSession]:
async with session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db),
) -> User:
payload = decode_token(token)
user_id = UUID(payload["sub"])
user = await db.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
return user
ルートハンドラー内でインラインにセッション、クライアント、または認証情報を作成しないでください。
非同期エンドポイント
I/Oを実行する場合はルートハンドラーを非同期にし、その内部で非同期ライブラリを使用します。
from fastapi import APIRouter, Depends, Query
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_current_user, get_db
from app.models.user import User
from app.schemas.user import UserResponse
router = APIRouter()
@router.get("/", response_model=list[UserResponse])
async def list_users(
limit: int = Query(default=50, ge=1, le=100),
offset: int = Query(default=0, ge=0),
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
result = await db.execute(
select(User).order_by(User.created_at.desc()).limit(limit).offset(offset)
)
return result.scalars().all()
非同期ハンドラーからの外部HTTP呼び出しにはhttpx.AsyncClientを使用してください。非同期ルートでrequestsを呼び出さないでください。
エラー処理
ドメイン例外を一元化し、レスポンスの形状を安定させます。
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
class ApiError(Exception):
def __init__(self, status_code: int, code: str, message: str):
self.status_code = status_code
self.code = code
self.message = message
def register_exception_handlers(app: FastAPI) -> None:
@app.exception_handler(ApiError)
async def api_error_handler(request: Request, exc: ApiError):
return JSONResponse(
status_code=exc.status_code,
content={"error": {"code": exc.code, "message": exc.message}},
)
OpenAPIカスタマイズ
カスタムOpenAPI呼び出し可能オブジェクトをapp.openapiに割り当ててください; 関数を一度だけ呼び出さないでください。
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
def install_openapi(app: FastAPI) -> None:
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
app.openapi_schema = get_openapi(
title="Service API",
version="1.0.0",
routes=app.routes,
)
return app.openapi_schema
app.openapi = custom_openapi
テスト
ルートハンドラーが決して参照しない内部ヘルパーではなく、Dependsで使用される依存関係をオーバーライドします。
import pytest
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies import get_db
from app.main import create_app
@pytest.fixture
async def client(test_session: AsyncSession):
app = create_app()
async def override_get_db():
yield test_session
app.dependency_overrides[get_db] = override_get_db
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test",
) as test_client:
yield test_client
app.dependency_overrides.clear()
セキュリティチェックリスト
argon2-cffi、bcrypt、または現在のpasslib互換ハッシャーでパスワードをハッシュする。- JWTの発行者、オーディエンス、有効期限、署名アルゴリズムを検証する。
- CORSオリジンを環境固有に保つ。
- 認証と書き込み負荷の高いエンドポイントにレート制限を設ける。
- すべてのリクエストボディにPydanticモデルを使用する。
- ORMパラメーターバインディングまたはSQLAlchemy Coreの式を使用する; f文字列でSQLを構築しない。
- ログからトークン、認可ヘッダー、クッキー、パスワードを削除する。
- CIで依存関係の監査ツールを実行する。
パフォーマンスチェックリスト
- データベース接続プールを明示的に設定する。
- リストエンドポイントにページネーションを追加する。
- N+1クエリに注意し、イーガーローディングを意図的に使用する。
- 非同期パスでは非同期HTTP/データベースクライアントを使用する。
- ペイロードサイズとCPUのトレードオフを確認してから圧縮を追加する。
- 明示的な無効化の後ろで安定した高コストの読み取りをキャッシュする。
使用例
これらの例はプロジェクト全体のテンプレートではなく、パターンとして使用してください:
- アプリケーションファクトリー:
create_appでミドルウェアとルーターを一度設定する。 - スキーマの分割:
UserCreate、UserUpdate、UserResponseはそれぞれ異なる責務を持つ。 - 依存関係のオーバーライド: テストは
get_dbを直接オーバーライドする。 - OpenAPIのカスタマイズ:
app.openapi = custom_openapiを割り当てる。
関連情報
- エージェント:
fastapi-reviewer - コマンド:
/fastapi-review - スキル:
python-patterns - スキル:
python-testing - スキル:
api-design
Source: affaan-m/ECC → docs/ja-JP/skills/fastapi-patterns/SKILL.md