mcpField
F
fastapi

fastapi

github.com/srbhr開発29.2k179 回インストール4 日前に更新
リポジトリ

fastapi について

Pydantic v2 と非同期 SQLAlchemy 2.0 を使う FastAPI 開発者向けのコーディングスキルです。API の実装や保守に役立つ構成例、実装パターン、既知の問題への対処をまとめています。

主な機能

  1. 01Pydantic スキーマと検証パターン
  2. 02非同期 SQLAlchemy のデータベース構成
  3. 03JWT 認証の実装例
  4. 04ドメイン別のプロジェクト構成
  5. 05既知の問題を避けるための指針

活用シーン

  1. 01FastAPI API プロジェクトの開始
  2. 02ドメイン別のコード構成
  3. 03JWT 保護ルートの追加
  4. 04検証や非同期処理の一般的なエラー回避

インストール

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

$npx skills add https://github.com/srbhr/resume-matcher --skill fastapi

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

詳細

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:フォームデータでField Setのメタデータが失われる

エラー: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)
    )

ルール:仕組みはどちらか一方を選び、統一してください。注入されたBackgroundTasksとResponse(background=...)を混在させないでください。

問題 #3:TestClientでOptionalフォームフィールドの検証に失敗する(リグレッション)

エラー:オプションの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: Annotated と ForwardRef の組み合わせで 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]

コメント