FastAPI 完全ガイド【2026年最新】｜Python 3.13 対応・Pydantic v2・async/await・Dependency Injection・OpenAPI 自動生成・SQLAlchemy 2.0 Async・OAuth2/JWT 認証・テスト・本番デプロイまでを実戦パターンで解説

# uv（2025 年以降の Python 標準マネージャ候補）
uv init my-api && cd my-api
uv add "fastapi[standard]>=0.130"
uv add "sqlalchemy[asyncio]>=2.0" "asyncpg>=0.30" "alembic>=1.13"
uv add --dev "pytest" "httpx" "ruff"

# 従来の pip でも同等
python -m venv .venv && source .venv/bin/activate
pip install "fastapi[standard]>=0.130" "sqlalchemy[asyncio]>=2.0" asyncpg alembic
pip install --upgrade pip

# 実行（開発モード: 自動リロード付き）
fastapi dev app/main.py

# 本番モード
fastapi run app/main.py

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    version="1.0.0",
    description="FastAPI with Python 3.13",
)

@app.get("/")
async def root() -> dict[str, str]:
    return {"message": "hello fastapi"}

@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok"}

fastapi dev app/main.py
#  Serving at: http://127.0.0.1:8000
#  API docs: http://127.0.0.1:8000/docs

# OpenAPI は自動生成
curl http://127.0.0.1:8000/openapi.json | jq .

from fastapi import FastAPI, status
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
    id: int
    title: str
    body: str

# 擬似 DB
posts: dict[int, Post] = {}

@app.get("/posts", response_model=list[Post])
async def list_posts() -> list[Post]:
    return list(posts.values())

@app.get("/posts/{post_id}", response_model=Post)
async def get_post(post_id: int) -> Post:
    return posts[post_id]

@app.post("/posts", response_model=Post, status_code=status.HTTP_201_CREATED)
async def create_post(post: Post) -> Post:
    posts[post.id] = post
    return post

@app.put("/posts/{post_id}", response_model=Post)
async def update_post(post_id: int, post: Post) -> Post:
    posts[post_id] = post
    return post

@app.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(post_id: int) -> None:
    posts.pop(post_id, None)

from fastapi import APIRouter

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

@router.get("/")
async def list_posts():
    return []

@router.get("/{post_id}")
async def get_post(post_id: int):
    return {"id": post_id}

from fastapi import FastAPI
from app.api import posts, users, auth

app = FastAPI()
app.include_router(posts.router)
app.include_router(users.router)
app.include_router(auth.router, prefix="/auth", tags=["auth"])

from typing import Annotated
from fastapi import FastAPI, Path, Query, Body, Header, Cookie
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: Annotated[str, Field(min_length=1, max_length=100)]
    price: Annotated[float, Field(gt=0, description="0 より大きい")]
    tags: list[str] = []

@app.post("/items/{category}")
async def create_item(
    # Path パラメータ: 正規表現・範囲・例
    category: Annotated[str, Path(pattern="^[a-z-]+$", examples=["books", "toys"])],

    # Query パラメータ: 範囲・別名・複数値
    q: Annotated[str | None, Query(min_length=1, max_length=50)] = None,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    tags: Annotated[list[str] | None, Query(alias="tag")] = None,

    # Body: Pydantic モデル
    item: Annotated[Item, Body()] = ...,

    # Header: HTTP ヘッダ（アンダースコア → ハイフン変換）
    x_request_id: Annotated[str | None, Header()] = None,

    # Cookie
    session: Annotated[str | None, Cookie()] = None,
):
    return {
        "category": category, "q": q, "limit": limit,
        "tags": tags, "item": item,
        "x_request_id": x_request_id, "session": session,
    }

from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, EmailStr, Field, ConfigDict, field_validator

class UserBase(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        json_schema_extra={
            "examples": [{"email": "alice@example.com", "name": "Alice"}],
        },
    )
    email: EmailStr
    name: Annotated[str, Field(min_length=1, max_length=50)]

class UserCreate(UserBase):
    password: Annotated[str, Field(min_length=8, max_length=72)]

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if v.isalpha() or v.isdigit():
            raise ValueError("英数字混在が必要")
        return v

class UserRead(UserBase):
    id: int
    created_at: datetime
    # SQLAlchemy モデルからの変換を許可（旧 orm_mode=True）
    model_config = ConfigDict(from_attributes=True)

from typing import Annotated
from fastapi import Depends, Header, HTTPException, status

async def require_api_key(
    x_api_key: Annotated[str | None, Header()] = None,
) -> str:
    if not x_api_key:
        raise HTTPException(status_code=401, detail="X-API-Key missing")
    if x_api_key != "expected":
        raise HTTPException(status_code=403, detail="invalid key")
    return x_api_key

# 型エイリアスでさらに簡潔に
ApiKey = Annotated[str, Depends(require_api_key)]

@app.get("/secure")
async def secure_endpoint(api_key: ApiKey):
    return {"api_key": api_key}

class Pagination:
    def __init__(
        self,
        page: Annotated[int, Query(ge=1)] = 1,
        size: Annotated[int, Query(ge=1, le=100)] = 20,
    ):
        self.page = page
        self.size = size
        self.offset = (page - 1) * size

PaginationDep = Annotated[Pagination, Depends()]

@app.get("/items")
async def list_items(pg: PaginationDep):
    return {"page": pg.page, "size": pg.size, "offset": pg.offset}

from fastapi import FastAPI, Depends

async def log_request():
    print("request received")

# アプリ全体に適用
app = FastAPI(dependencies=[Depends(log_request)])

# 特定ルーターにだけ適用も可能
# router = APIRouter(dependencies=[Depends(require_api_key)])

from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

SECRET_KEY = "change-me-in-env"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

class TokenPayload(BaseModel):
    sub: str
    exp: datetime

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

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

def create_access_token(subject: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"sub": subject, "exp": expire}
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
) -> str:
    credentials_exc = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="invalid token",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        data = TokenPayload(**jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]))
    except JWTError:
        raise credentials_exc
    return data.sub

CurrentUser = Annotated[str, Depends(get_current_user)]

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from typing import Annotated

from app.core.security import (
    verify_password, create_access_token, CurrentUser,
)

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

@router.post("/token")
async def login(
    form: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> dict[str, str]:
    # form.username / form.password を DB 照合（例示）
    user = fake_users_db.get(form.username)
    if not user or not verify_password(form.password, user["hashed_password"]):
        raise HTTPException(status_code=401, detail="invalid credentials")
    return {
        "access_token": create_access_token(subject=form.username),
        "token_type": "bearer",
    }

@router.get("/me")
async def me(user: CurrentUser) -> dict[str, str]:
    return {"username": user}

from fastapi import BackgroundTasks

async def send_welcome_mail(email: str, name: str) -> None:
    # 実際は SMTP やサードパーティ API を呼ぶ
    print(f"send to {email} ({name})")

@app.post("/signup")
async def signup(
    email: str,
    name: str,
    bg: BackgroundTasks,
):
    # DB 保存など同期で終わらせる
    user = await db.user.create(email=email, name=name)
    # レスポンス送信後に実行される
    bg.add_task(send_welcome_mail, email=email, name=name)
    return {"id": user.id}

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    def __init__(self) -> None:
        self.active: list[WebSocket] = []

    async def connect(self, ws: WebSocket) -> None:
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket) -> None:
        self.active.remove(ws)

    async def broadcast(self, message: str) -> None:
        for ws in list(self.active):
            await ws.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def chat(ws: WebSocket) -> None:
    await manager.connect(ws)
    try:
        while True:
            msg = await ws.receive_text()
            await manager.broadcast(msg)
    except WebSocketDisconnect:
        manager.disconnect(ws)

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://app.example.com",
        "http://localhost:3000",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

import uuid
from fastapi import Request

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = request.headers.get("x-request-id") or uuid.uuid4().hex
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["x-request-id"] = request_id
    return response

from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine

engine = None   # 後で格納

@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup
    global engine
    engine = create_async_engine("postgresql+asyncpg://user:pw@db/app")
    print("DB connected")
    yield
    # shutdown
    await engine.dispose()
    print("DB closed")

app = FastAPI(lifespan=lifespan)

from sqlalchemy.ext.asyncio import (
    AsyncSession, async_sessionmaker, create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:pw@localhost:5432/app"

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

class Base(DeclarativeBase):
    pass

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

from datetime import datetime
from sqlalchemy import String, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db.database import Base

class User(Base):
    __tablename__ = "users"

    id:         Mapped[int]       = mapped_column(primary_key=True)
    email:      Mapped[str]       = mapped_column(String(255), unique=True, index=True)
    name:       Mapped[str]       = mapped_column(String(50))
    hashed_pw:  Mapped[str]       = mapped_column(String(72))
    created_at: Mapped[datetime]  = mapped_column(DateTime(timezone=True), server_default=func.now())

from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.database import get_db
from app.db.models import User
from app.schemas.user import UserCreate, UserRead
from app.core.security import hash_password

router = APIRouter(prefix="/users", tags=["users"])
DB = Annotated[AsyncSession, Depends(get_db)]

@router.post("/", response_model=UserRead, status_code=201)
async def create_user(payload: UserCreate, db: DB):
    user = User(
        email=payload.email,
        name=payload.name,
        hashed_pw=hash_password(payload.password),
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@router.get("/", response_model=list[UserRead])
async def list_users(db: DB):
    result = await db.execute(select(User).order_by(User.id))
    return result.scalars().all()

@router.get("/{user_id}", response_model=UserRead)
async def get_user(user_id: int, db: DB):
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="not found")
    return user

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_root():
    r = client.get("/")
    assert r.status_code == 200
    assert r.json() == {"message": "hello fastapi"}

def test_create_item():
    r = client.post("/posts", json={"id": 1, "title": "t", "body": "b"})
    assert r.status_code == 201
    data = r.json()
    assert data["title"] == "t"

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.mark.asyncio
async def test_async_endpoint():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        r = await ac.get("/users/")
        assert r.status_code == 200

import pytest_asyncio
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

from app.db.database import Base, get_db
from app.main import app

TEST_DB_URL = "postgresql+asyncpg://test:test@localhost:5432/test"

@pytest_asyncio.fixture
async def db_session():
    engine = create_async_engine(TEST_DB_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    Session = async_sessionmaker(engine, expire_on_commit=False)

    async def override_get_db():
        async with Session() as s:
            yield s

    app.dependency_overrides[get_db] = override_get_db
    yield
    app.dependency_overrides.clear()
    await engine.dispose()

# バックエンド起動中に openapi-typescript で取得
npx openapi-typescript http://localhost:8000/openapi.json -o frontend/src/api/schema.ts

# tRPC 風に使うなら openapi-fetch
# https://openapi-ts.dev/openapi-fetch/

# 開発
fastapi dev app/main.py

# 本番（--workers = CPU コア数 x 2 + 1 が目安）
fastapi run app/main.py --workers 4 --host 0.0.0.0 --port 8000

# systemd 化するならサービスファイルを用意
# /etc/systemd/system/my-api.service
# [Service]
# ExecStart=/opt/venv/bin/fastapi run /opt/app/main.py --workers 4
# Restart=always

# syntax=docker/dockerfile:1.7
FROM python:3.13-slim AS base

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1 \
    PIP_NO_CACHE_DIR=1

WORKDIR /app

# 依存のみ先に入れてレイヤキャッシュを効かせる
COPY pyproject.toml uv.lock ./
RUN pip install uv && uv sync --frozen --no-dev

COPY app app

EXPOSE 8000
CMD ["uv", "run", "fastapi", "run", "app/main.py", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

services:
  api:
    build: .
    ports: ["8000:8000"]
    environment:
      DATABASE_URL: postgresql+asyncpg://app:pw@db:5432/app
      SECRET_KEY: ${SECRET_KEY}
    depends_on: [db]

  db:
    image: postgres:17-alpine
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: pw
      POSTGRES_DB: app
    volumes: ["pgdata:/var/lib/postgresql/data"]
    ports: ["5432:5432"]

volumes:
  pgdata:

# fastapi[standard] に同梱
fastapi deploy
# ログイン → プロジェクト選択 → 自動デプロイ
# 環境変数・ドメイン・SSL まで自動

my-api/
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── .env.example
├── alembic.ini
├── migrations/                 # Alembic
├── app/
│   ├── main.py                 # FastAPI インスタンス & ルーター include
│   ├── core/
│   │   ├── config.py           # pydantic-settings
│   │   ├── security.py         # JWT / パスワードハッシュ
│   │   └── logging.py
│   ├── db/
│   │   ├── database.py         # engine / session
│   │   ├── models.py           # SQLAlchemy 2.0 モデル
│   │   └── base.py
│   ├── schemas/                # Pydantic v2 モデル（入出力 DTO）
│   │   ├── user.py
│   │   └── post.py
│   ├── api/
│   │   ├── deps.py             # 共通依存性
│   │   ├── auth.py
│   │   ├── users.py
│   │   └── posts.py
│   ├── services/               # ビジネスロジック（ルーターと DB の中間）
│   └── worker/                 # Celery / arq タスク
└── tests/
    ├── conftest.py
    ├── test_auth.py
    └── test_users.py