mcpField
F
fastapi

fastapi

github.com/jezweb開発1.1k2.0k 回インストール3 週間前に更新
リポジトリ

fastapi について

fastapi は、FastAPI 開発者向けの実践ガイドです。Pydantic v2、非同期 SQLAlchemy 2.0、JWT認証のパターンに加え、既知の問題や一般的なエラーの対処法も扱います。

主な機能

  1. 01プロジェクト構成とルーターの整理
  2. 02Pydantic v2によるデータ検証
  3. 03非同期SQLAlchemy 2.0データベースの利用
  4. 04JWT認証とルート保護の実装
  5. 05既知の問題と一般的なエラーへの対処

活用シーン

  1. 01FastAPIプロジェクトの基本構成の作成
  2. 02非同期データベースを使う検証付きAPIの構築
  3. 03APIへのJWTログインとアクセス制御の追加
  4. 04フォーム、OpenAPI、非同期処理の問題の調査

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/jezweb/claude-skills --skill fastapi

インストール後、Claude Code で/fastapiを入力して読み込まれたか確認します。

同梱ファイル

SKILL.md と一緒にインストールされる参考資料やスクリプト。

README.md
.claude-plugin/
plugin.json
templates/
.env.example
pyproject.toml
templates/src/
__init__.py
config.py
database.py
main.py
templates/src/auth/
__init__.py
dependencies.py
models.py
router.py
schemas.py
service.py
templates/tests/
__init__.py
test_main.py

詳細

FastAPI スキル

Pydantic v2、SQLAlchemy 2.0 async、JWT 認証を使用した、本番環境で検証済みの FastAPI パターンです。

最新バージョン(2026年1月確認済み):

  • FastAPI:0.128.0
  • Pydantic:2.11.7
  • SQLAlchemy:2.0.30
  • Uvicorn:0.35.0
  • python-jose:3.3.0

要件:

  • Python 3.9 以降(FastAPI 0.125.0 で Python 3.8 のサポートは終了しました)
  • Pydantic v2.7.0 以降(FastAPI 0.128.0 で Pydantic v1 のサポートは完全に削除されました)

クイックスタート

uv を使用したプロジェクトのセットアップ
# Create project
uv init my-api
cd my-api

# Add dependencies
uv add fastapi[standard] sqlalchemy[asyncio] aiosqlite python-jose[cryptography] passlib[bcrypt]

# Run development server
uv run fastapi dev src/main.py
最小限の実行可能な例
# src/main.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="My API")

class Item(BaseModel):
    name: str
    price: float

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.post("/items")
async def create_item(item: Item):
    return item

実行:uv run fastapi dev src/main.py

ドキュメントはこちらで確認できます:http://127.0.0.1:8000/docs


プロジェクト構成(ドメインベース)

保守しやすいプロジェクトにするには、ファイルの種類ではなくドメインごとに整理します。

my-api/
├── pyproject.toml
├── src/
│   ├── __init__.py
│   ├── main.py              # FastAPI app initialization
│   ├── config.py            # Global settings
│   ├── database.py          # Database connection
│   │
│   ├── auth/                # Auth domain
│   │   ├── __init__.py
│   │   ├── router.py        # Auth endpoints
│   │   ├── schemas.py       # Pydantic models
│   │   ├── models.py        # SQLAlchemy models
│   │   ├── service.py       # Business logic
│   │   └── dependencies.py  # Auth dependencies
│   │
│   ├── items/               # Items domain
│   │   ├── __init__.py
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   └── service.py
│   │
│   └── shared/              # Shared utilities
│       ├── __init__.py
│       └── exceptions.py
└── tests/
    └── test_main.py

コアパターン

Pydantic スキーマ(バリデーション)
# src/items/schemas.py
from pydantic import BaseModel, Field, ConfigDict
from datetime import datetime
from enum import Enum

class ItemStatus(str, Enum):
    DRAFT = "draft"
    PUBLISHED = "published"
    ARCHIVED = "archived"

class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str | None = Field(None, max_length=500)
    price: float = Field(..., gt=0, description="Price must be positive")
    status: ItemStatus = ItemStatus.DRAFT

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: str | None = Field(None, min_length=1, max_length=100)
    description: str | None = None
    price: float | None = Field(None, gt=0)
    status: ItemStatus | None = None

class ItemResponse(ItemBase):
    id: int
    created_at: datetime

    model_config = ConfigDict(from_attributes=True)

重要なポイント:

  • バリデーション制約には Field() を使用します
  • Create/Update/Response スキーマを分けます
  • from_attributes=True により、SQLAlchemy モデルを変換できます
  • Optional[str] ではなく str | None(Python 3.10 以降)を使用します
SQLAlchemy モデル(データベース)
# src/items/models.py
from sqlalchemy import String, Float, DateTime, Enum as SQLEnum
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime
from src.database import Base
from src.items.schemas import ItemStatus

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    description: Mapped[str | None] = mapped_column(String(500), nullable=True)
    price: Mapped[float] = mapped_column(Float)
    status: Mapped[ItemStatus] = mapped_column(
        SQLEnum(ItemStatus), default=ItemStatus.DRAFT
    )
    created_at: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )
データベースのセットアップ(Async SQLAlchemy 2.0)
# src/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "sqlite+aiosqlite:///./database.db"

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
ルーターパターン
# src/items/router.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.items import schemas, models

router = APIRouter(prefix="/items", tags=["items"])

@router.get("", response_model=list[schemas.ItemResponse])
async def list_items(
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(models.Item).offset(skip).limit(limit)
    )
    return result.scalars().all()

@router.get("/{item_id}", response_model=schemas.ItemResponse)
async def get_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(
        select(models.Item).where(models.Item.id == item_id)
    )
    item = result.scalar_one_or_none()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.post("", response_model=schemas.ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(
    item_in: schemas.ItemCreate,
    db: AsyncSession = Depends(get_db)
):
    item = models.Item(**item_in.model_dump())
    db.add(item)
    await db.commit()
    await db.refresh(item)
    return item
メインアプリ
# src/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from src.database import engine, Base
from src.items.router import router as items_router
from src.auth.router import router as auth_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Create tables
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown: cleanup if needed

app = FastAPI(title="My API", lifespan=lifespan)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Your frontend
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Include routers
app.include_router(auth_router)
app.include_router(items_router)

JWT 認証

認証スキーマ
# src/auth/schemas.py
from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserResponse(BaseModel):
    id: int
    email: str

    model_config = ConfigDict(from_attributes=True)

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    user_id: int | None = None
認証サービス
# src/auth/service.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from src.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")

def decode_token(token: str) -> dict | None:
    try:
        return jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    except JWTError:
        return None
認証依存関係
# src/auth/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.auth import service, models, schemas

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
) -> models.User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

    payload = service.decode_token(token)
    if payload is None:
        raise credentials_exception

    user_id = payload.get("sub")
    if user_id is None:
        raise credentials_exception

    result = await db.execute(
        select(models.User).where(models.User.id == int(user_id))
    )
    user = result.scalar_one_or_none()

    if user is None:
        raise credentials_exception

    return user
認証ルーター
# src/auth/router.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from src.database import get_db
from src.auth import schemas, models, service
from src.auth.dependencies import get_current_user

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register", response_model=schemas.UserResponse)
async def register(
    user_in: schemas.UserCreate,
    db: AsyncSession = Depends(get_db)
):
    # Check existing
    result = await db.execute(
        select(models.User).where(models.User.email == user_in.email)
    )
    if result.scalar_one_or_none():
        raise HTTPException(status_code=400, detail="Email already registered")

    user = models.User(
        email=user_in.email,
        hashed_password=service.hash_password(user_in.password)
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@router.post("/login", response_model=schemas.Token)
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(models.User).where(models.User.email == form_data.username)
    )
    user = result.scalar_one_or_none()

    if not user or not service.verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    access_token = service.create_access_token(data={"sub": str(user.id)})
    return schemas.Token(access_token=access_token)

@router.get("/me", response_model=schemas.UserResponse)
async def get_me(current_user: models.User = Depends(get_current_user)):
    return current_user
ルートを保護する
# In any router
from src.auth.dependencies import get_current_user
from src.auth.models import User

@router.post("/items")
async def create_item(
    item_in: schemas.ItemCreate,
    current_user: User = Depends(get_current_user),  # Requires auth
    db: AsyncSession = Depends(get_db)
):
    item = models.Item(**item_in.model_dump(), user_id=current_user.id)
    # ...

設定

# src/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str = "sqlite+aiosqlite:///./database.db"
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

.env を作成します:

DATABASE_URL=sqlite+aiosqlite:///./database.db
SECRET_KEY=your-super-secret-key-here
ACCESS_TOKEN_EXPIRE_MINUTES=30

重要なルール

必ず行うこと
  1. Pydantic スキーマと SQLAlchemy モデルを分ける - 役割が異なるため、ファイルも分けます
  2. I/O 操作には async を使用する - データベース、HTTP 呼び出し、ファイルアクセス
  3. Pydantic の Field() で検証する - 制約、デフォルト値、説明
  4. 依存性注入を使用する - データベース、認証、バリデーションには Depends() を使用します
  5. 適切なステータスコードを返す - 作成には 201、削除には 204 など
絶対に行わないこと
  1. async ルート内でブロッキング呼び出しを使用しない - time.sleep() は使わず、asyncio.sleep() を使用します
  2. ルートにビジネスロジックを記述しない - サービスレイヤーを使用します
  3. シークレットをハードコードしない - 環境変数を使用します
  4. バリデーションを省略しない - 必ず Pydantic スキーマを使用します
  5. 本番環境で CORS オリジンに * を使用しない - オリジンを正確に指定します

既知の問題の防止

このスキルは、公式 FastAPI の GitHub およびリリースノートに記録されている 7 件の問題を防止します。

問題 #1:フォームデータでフィールドセットのメタデータが失われる

エラー:Form() の使用時に model.model_fields_set にデフォルト値が含まれる 出典:GitHub Issue #13399 発生する理由:フォームデータの解析でデフォルト値が事前に読み込まれてバリデーターに渡されるため、ユーザーが明示的に設定したフィールドとデフォルト値が使われたフィールドを区別できなくなります。このバグはフォームデータにのみ影響し、JSON ボディデータには影響しません。

防止策:

# ✗ AVOID: Pydantic model with Form when you need field_set metadata
from typing import Annotated
from fastapi import Form

@app.post("/form")
async def endpoint(model: Annotated[MyModel, Form()]):
    fields = model.model_fields_set  # Unreliable! ❌

# ✓ USE: Individual form fields or JSON body instead
@app.post("/form-individual")
async def endpoint(
    field_1: Annotated[bool, Form()] = True,
    field_2: Annotated[str | None, Form()] = None
):
    # You know exactly what was provided ✓

# ✓ OR: Use JSON body when metadata matters
@app.post("/json")
async def endpoint(model: MyModel):
    fields = model.model_fields_set  # Works correctly ✓
問題 #2:カスタム Response により BackgroundTasks が通知なく上書きされる

エラー:BackgroundTasks 依存関係を介して追加したバックグラウンドタスクが実行されない 出典:GitHub Issue #11215 発生する理由:background パラメーターを指定したカスタム Response を返すと、注入された BackgroundTasks 依存関係に追加されたすべてのタスクが上書きされます。これはドキュメントに記載されておらず、エラーが通知されないまま発生します。

防止策:

# ✗ WRONG: Mixing both mechanisms
from fastapi import BackgroundTasks
from starlette.responses import Response, BackgroundTask

@app.get("/")
async def endpoint(tasks: BackgroundTasks):
    tasks.add_task(send_email)  # This will be lost! ❌
    return Response(
        content="Done",
        background=BackgroundTask(log_event)  # Only this runs
    )

# ✓ RIGHT: Use only BackgroundTasks dependency
@app.get("/")
async def endpoint(tasks: BackgroundTasks):
    tasks.add_task(send_email)
    tasks.add_task(log_event)
    return {"status": "done"}  # All tasks run ✓

# ✓ OR: Use only Response background (but can't inject dependencies)
@app.get("/")
async def endpoint():
    return Response(
        content="Done",
        background=BackgroundTask(log_event)
    )

ルール:仕組みを 1 つだけ選び、一貫して使用します。注入された BackgroundTasks と Response(background=...) を併用しないでください。

問題 #3:TestClient でオプションのフォームフィールドが機能しない(リグレッション)

エラー:オプションの Literal フィールドで 422: "Input should be 'abc' or 'def'" が発生する 出典:GitHub Issue #12245 発生する理由:FastAPI 0.114.0 以降、Literal 型のオプションのフォームフィールドに TestClient 経由で None を渡すと、バリデーションに失敗します。0.113.0 では正常に動作していました。

防止策:

from typing import Annotated, Literal, Optional
from fastapi import Form
from fastapi.testclient import TestClient

# ✗ PROBLEMATIC: Optional Literal with Form (breaks in 0.114.0+)
@app.post("/")
async def endpoint(
    attribute: Annotated[Optional[Literal["abc", "def"]], Form()]
):
    return {"attribute": attribute}

client = TestClient(app)
data = {"attribute": None}  # or omit the field
response = client.post("/", data=data)  # Returns 422 ❌

# ✓ WORKAROUND 1: Don't pass None explicitly, omit the field
data = {}  # Omit instead of None
response = client.post("/", data=data)  # Works ✓

# ✓ WORKAROUND 2: Avoid Literal types with optional form fields
@app.post("/")
async def endpoint(attribute: Annotated[str | None, Form()] = None):
    # Validate in application logic instead
    if attribute and attribute not in ["abc", "def"]:
        raise HTTPException(400, "Invalid attribute")
Issue #4:Pydantic の Json 型がフォームデータで機能しない

エラー:"JSON object must be str, bytes or bytearray" ソース:GitHub Issue #10997 原因:Pydantic の Json 型を Form() と直接組み合わせると失敗します。フィールドを str として受け取り、手動で解析する必要があります。

予防策:

from typing import Annotated
from fastapi import Form
from pydantic import Json, BaseModel

# ✗ WRONG: Json type directly with Form
@app.post("/broken")
async def broken(json_list: Annotated[Json[list[str]], Form()]) -> list[str]:
    return json_list  # Returns 422 ❌

# ✓ RIGHT: Accept as str, parse with Pydantic
class JsonListModel(BaseModel):
    json_list: Json[list[str]]

@app.post("/working")
async def working(json_list: Annotated[str, Form()]) -> list[str]:
    model = JsonListModel(json_list=json_list)  # Pydantic parses here
    return model.json_list  # Works ✓
Issue #5:ForwardRef を使った Annotated で OpenAPI の生成に失敗する

エラー:依存関係の型に対する OpenAPI スキーマがない、または不正である ソース:GitHub Issue #13056 原因:Annotated を Depends() と前方参照(__future__ import annotations によるもの)と組み合わせると、OpenAPI スキーマの生成に失敗するか、不正なスキーマが生成されます。

予防策:

# ✗ PROBLEMATIC: Forward reference with Depends
from __future__ import annotations
from dataclasses import dataclass
from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

def get_potato() -> Potato:  # Forward reference
    return Potato(color='red', size=10)

@app.get('/')
async def read_root(potato: Annotated[Potato, Depends(get_potato)]):
    return {'Hello': 'World'}
# OpenAPI schema doesn't include Potato definition correctly ❌

@dataclass
class Potato:
    color: str
    size: int

# ✓ WORKAROUND 1: Don't use __future__ annotations in route files
# Remove: from __future__ import annotations

# ✓ WORKAROUND 2: Use string literals for type hints
def get_potato() -> "Potato":
    return Potato(color='red', size=10)

# ✓ WORKAROUND 3: Define classes before they're used in dependencies
@dataclass
class Potato:
    color: str
    size: int

def get_potato() -> Potato:  # Now works ✓
    return Potato(color='red', size=10)
Issue #6:Pydantic v2 のパスパラメーターにおける Union 型の破壊的変更

エラー:Pydantic v2 では、int | str 型のパスパラメーターが常に str として解析される ソース:GitHub Issue #11251 | コミュニティ提供 原因:Pydantic v1 から v2 への移行に伴う重大な破壊的変更です。パス/クエリパラメーターに str を含む Union 型を指定すると、常に str として解析されるようになりました(v1 では正しく動作していました)。

予防策:

from uuid import UUID

# ✗ PROBLEMATIC: Union with str in path parameter
@app.get("/int/{path}")
async def int_path(path: int | str):
    return str(type(path))
    # Pydantic v1: returns <class 'int'> for "123"
    # Pydantic v2: returns <class 'str'> for "123" ❌

@app.get("/uuid/{path}")
async def uuid_path(path: UUID | str):
    return str(type(path))
    # Pydantic v1: returns <class 'uuid.UUID'> for valid UUID
    # Pydantic v2: returns <class 'str'> ❌

# ✓ RIGHT: Avoid union types with str in path/query parameters
@app.get("/int/{path}")
async def int_path(path: int):
    return str(type(path))  # Works correctly ✓

# ✓ ALTERNATIVE: Use validators if type coercion needed
from pydantic import field_validator

class PathParams(BaseModel):
    path: int | str

    @field_validator('path')
    def coerce_to_int(cls, v):
        if isinstance(v, str) and v.isdigit():
            return int(v)
        return v
Issue #7:field_validator 内の ValueError が 422 ではなく 500 を返す

エラー:カスタムバリデーターで ValueError を発生させると 500 Internal Server Error になる ソース:GitHub Discussion #10779 | コミュニティ提供 原因:Form フィールドを含む Pydantic の @field_validator 内で ValueError を発生させると、FastAPI は期待される 422 Unprocessable Entity のバリデーションエラーではなく、500 Internal Server Error を返します。

予防策:

from typing import Annotated
from fastapi import Form
from pydantic import BaseModel, field_validator, ValidationError, Field

# ✗ WRONG: ValueError in validator
class MyForm(BaseModel):
    value: int

    @field_validator('value')
    def validate_value(cls, v):
        if v < 0:
            raise ValueError("Value must be positive")  # Returns 500! ❌
        return v

# ✓ RIGHT 1: Raise ValidationError instead
class MyForm(BaseModel):
    value: int

    @field_validator('value')
    def validate_value(cls, v):
        if v < 0:
            raise ValidationError("Value must be positive")  # Returns 422 ✓
        return v

# ✓ RIGHT 2: Use Pydantic's built-in constraints
class MyForm(BaseModel):
    value: Annotated[int, Field(gt=0)]  # Built-in validation, returns 422 ✓

よくあるエラーと修正方法

422 Unprocessable Entity

原因:リクエストボディが Pydantic スキーマと一致していません。

デバッグ方法:

  1. /docs エンドポイントを確認し、まずそこでテストします
  2. JSON 構造がスキーマと一致していることを確認します
  3. 必須フィールドと任意フィールドを確認します

修正方法:カスタムバリデーションエラーハンドラーを追加します:

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body}
    )
CORS エラー

原因:CORS ミドルウェアがない、または設定が正しくありません。

修正方法:

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # Not "*" in production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
非同期イベントループのブロッキング

原因:非同期ルート内でのブロッキング呼び出し(例:time.sleep()、同期データベースクライアント、CPU 負荷の高い処理)

症状(本番環境規模):

  • スループットが予想よりはるかに早く頭打ちになる
  • 同時実行数が増えるにつれて、レイテンシーが「急増」する
  • 負荷がかかると、リクエストの処理パターンがほぼ直列に見える
  • イベントループが飽和すると、リクエストがいつまでもキューに滞留する
  • 無限ループではないものの、目立たない小さなブロッキング呼び出しが散在している

修正方法:非同期の代替手段を使用します:

# ✗ WRONG: Blocks event loop
import time
from sqlalchemy import create_engine  # Sync client

@app.get("/users")
async def get_users():
    time.sleep(0.1)  # Even small blocking adds up at scale!
    result = sync_db_client.query("SELECT * FROM users")  # Blocks!
    return result

# ✓ RIGHT 1: Use async database driver
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    await asyncio.sleep(0.1)  # Non-blocking
    result = await db.execute(select(User))
    return result.scalars().all()

# ✓ RIGHT 2: Use def (not async def) for CPU-bound routes
# FastAPI runs def routes in thread pool automatically
@app.get("/cpu-heavy")
def cpu_heavy_task():  # Note: def not async def
    return expensive_cpu_work()  # Runs in thread pool ✓

# ✓ RIGHT 3: Use run_in_executor for blocking calls in async routes
import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()

@app.get("/mixed")
async def mixed_task():
    # Run blocking function in thread pool
    result = await asyncio.get_event_loop().run_in_executor(
        executor,
        blocking_function  # Your blocking function
    )
    return result

ソース:本番環境での事例研究(2026年1月) | コミュニティ提供

任意フィールドで「Field required」が発生する

原因:デフォルト値なしで Optional[str] を使用しています。

修正方法:

# Wrong
description: Optional[str]  # Still required!

# Right
description: str | None = None  # Optional with default

テスト

# tests/test_main.py
import pytest
from httpx import AsyncClient, ASGITransport
from src.main import app

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_root(client):
    response = await client.get("/")
    assert response.status_code == 200

@pytest.mark.asyncio
async def test_create_item(client):
    response = await client.post(
        "/items",
        json={"name": "Test", "price": 9.99}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Test"

実行:uv run pytest


デプロイ

Uvicorn(開発環境)
uv run fastapi dev src/main.py
Uvicorn(本番環境)
uv run uvicorn src.main:app --host 0.0.0.0 --port 8000
Gunicorn + Uvicorn(ワーカーを使用した本番環境)
uv add gunicorn
uv run gunicorn src.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
Docker
FROM python:3.12-slim

WORKDIR /app
COPY . .

RUN pip install uv && uv sync

EXPOSE 8000
CMD ["uv", "run", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

参考資料


最終確認日:2026-01-21 | スキルのバージョン:1.1.0 | 変更内容:既知の問題を7件追加(フォームデータのバグ、バックグラウンドタスク、Pydantic v2 移行時の注意点)。本番環境のパターンを含め、非同期処理のブロッキングに関するガイダンスを拡充 メンテナー:Jezweb | [email protected]

コメント