Python バックエンド 2025 完全ガイド：FastAPI vs Django、非同期処理、AIサービス構築まで

FastAPIを選ぶ場合：
  - 新しいAPIサービスを構築する時
  - AI/MLモデルサービングが必要な時
  - 高い並行性が要求される時
  - マイクロサービスアーキテクチャの時

Djangoを選ぶ場合：
  - 管理パネルが必要な時
  - 迅速なCRUDアプリ開発が必要な時
  - フルスタックWebアプリケーションの時
  - チームにDjango経験が豊富な時

Flaskを選ぶ場合：
  - 極めてシンプルなAPIの時
  - レガシーシステムのメンテナンスの時
  - 学習目的の時

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from datetime import datetime

app = FastAPI(
    title="My API",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., pattern=r"^[\w\.\-]+@[\w\.\-]+\.\w+$")
    age: int = Field(..., ge=0, le=150)

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    created_at: datetime

    model_config = {"from_attributes": True}

@app.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    return UserResponse(
        id=1,
        username=user.username,
        email=user.email,
        created_at=datetime.now(),
    )

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
    if user_id <= 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )
    return UserResponse(id=user_id, username="test", email="t@t.com", created_at=datetime.now())

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional
from enum import Enum

class Role(str, Enum):
    ADMIN = "admin"
    USER = "user"
    MODERATOR = "moderator"

class UserProfile(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    role: Role = Role.USER
    tags: list[str] = Field(default_factory=list, max_length=10)
    metadata: Optional[dict[str, str]] = None

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Name cannot be whitespace only")
        return v.strip()

    @model_validator(mode="after")
    def validate_admin_tags(self):
        if self.role == Role.ADMIN and not self.tags:
            raise ValueError("Admin must have at least one tag")
        return self

    model_config = {
        "json_schema_extra": {
            "examples": [
                {"name": "Alice", "role": "admin", "tags": ["ops"]}
            ]
        }
    }

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

# DBセッション依存性
async def get_db() -> AsyncSession:
    async with async_session_maker() as session:
        try:
            yield session
        finally:
            await session.close()

# 認証依存性
async def get_current_user(
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db),
) -> User:
    token = authorization.replace("Bearer ", "")
    user = await verify_token(token, db)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# 権限検証依存性
def require_role(required_role: str):
    async def role_checker(user: User = Depends(get_current_user)):
        if user.role != required_role:
            raise HTTPException(status_code=403, detail="Insufficient permissions")
        return user
    return role_checker

# 使用例
@app.get("/admin/users")
async def list_users(
    admin: Annotated[User, Depends(require_role("admin"))],
    db: Annotated[AsyncSession, Depends(get_db)],
):
    return await get_all_users(db)

from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time
import logging

logger = logging.getLogger(__name__)

# CORSミドルウェア
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# カスタムミドルウェア — リクエストロギング
class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        elapsed = time.perf_counter() - start
        logger.info(
            f"method={request.method} path={request.url.path} "
            f"status={response.status_code} duration={elapsed:.3f}s"
        )
        return response

app.add_middleware(RequestLoggingMiddleware)

# グローバル例外ハンドラ
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=422,
        content={"detail": str(exc), "type": "validation_error"},
    )

from fastapi import WebSocket, WebSocketDisconnect, BackgroundTasks

# WebSocketチャット
class ConnectionManager:
    def __init__(self):
        self.active: list[WebSocket] = []

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

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

    async def broadcast(self, message: str):
        for conn in self.active:
            await conn.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def websocket_chat(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            data = await ws.receive_text()
            await manager.broadcast(f"User: {data}")
    except WebSocketDisconnect:
        manager.disconnect(ws)

# バックグラウンドタスク
async def send_notification(email: str, message: str):
    await asyncio.sleep(2)
    logger.info(f"Sent notification to {email}")

@app.post("/orders")
async def create_order(order: OrderCreate, background_tasks: BackgroundTasks):
    result = await save_order(order)
    background_tasks.add_task(send_notification, order.email, "Order confirmed!")
    return result

# models.py
from django.db import models
from django.core.validators import MinValueValidator

class Category(models.Model):
    name = models.CharField(max_length=100, unique=True)
    slug = models.SlugField(unique=True)

    class Meta:
        verbose_name_plural = "categories"
        ordering = ["name"]

    def __str__(self):
        return self.name

class Product(models.Model):
    name = models.CharField(max_length=200)
    category = models.ForeignKey(Category, on_delete=models.CASCADE, related_name="products")
    price = models.DecimalField(max_digits=10, decimal_places=2, validators=[MinValueValidator(0)])
    stock = models.PositiveIntegerField(default=0)
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        indexes = [
            models.Index(fields=["category", "-created_at"]),
            models.Index(fields=["price"]),
        ]

# クエリ例 — N+1問題の解決
products = Product.objects.select_related("category").filter(is_active=True)

# 集計
from django.db.models import Avg, Count, Q
stats = Category.objects.annotate(
    product_count=Count("products"),
    avg_price=Avg("products__price"),
    active_count=Count("products", filter=Q(products__is_active=True)),
)

# admin.py
from django.contrib import admin
from .models import Product, Category

@admin.register(Product)
class ProductAdmin(admin.ModelAdmin):
    list_display = ["name", "category", "price", "stock", "is_active", "created_at"]
    list_filter = ["is_active", "category", "created_at"]
    search_fields = ["name", "category__name"]
    list_editable = ["price", "stock", "is_active"]
    readonly_fields = ["created_at", "updated_at"]
    list_per_page = 25

    actions = ["activate_products", "deactivate_products"]

    @admin.action(description="Activate selected products")
    def activate_products(self, request, queryset):
        updated = queryset.update(is_active=True)
        self.message_user(request, f"{updated} products activated.")

    @admin.action(description="Deactivate selected products")
    def deactivate_products(self, request, queryset):
        updated = queryset.update(is_active=False)
        self.message_user(request, f"{updated} products deactivated.")

# serializers.py
from rest_framework import serializers
from .models import Product, Category

class CategorySerializer(serializers.ModelSerializer):
    product_count = serializers.IntegerField(read_only=True)

    class Meta:
        model = Category
        fields = ["id", "name", "slug", "product_count"]

class ProductSerializer(serializers.ModelSerializer):
    category_name = serializers.CharField(source="category.name", read_only=True)

    class Meta:
        model = Product
        fields = ["id", "name", "category", "category_name", "price", "stock", "is_active"]

# views.py
from rest_framework import viewsets, filters
from django_filters.rest_framework import DjangoFilterBackend

class ProductViewSet(viewsets.ModelViewSet):
    queryset = Product.objects.select_related("category").filter(is_active=True)
    serializer_class = ProductSerializer
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_fields = ["category", "is_active"]
    search_fields = ["name"]
    ordering_fields = ["price", "created_at"]
    ordering = ["-created_at"]

# Django 5.0+ 非同期ビュー
from django.http import JsonResponse
from asgiref.sync import sync_to_async

async def async_product_list(request):
    # ORM呼び出しはsync_to_asyncでラップが必要
    products = await sync_to_async(list)(
        Product.objects.select_related("category").filter(is_active=True)[:20]
    )
    data = [
        {"id": p.id, "name": p.name, "price": str(p.price)}
        for p in products
    ]
    return JsonResponse({"products": data})

# Django Channels — WebSocket
from channels.generic.websocket import AsyncJsonWebSocketConsumer

class ChatConsumer(AsyncJsonWebSocketConsumer):
    async def connect(self):
        self.room = self.scope["url_route"]["kwargs"]["room"]
        await self.channel_layer.group_add(self.room, self.channel_name)
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(self.room, self.channel_name)

    async def receive_json(self, content):
        await self.channel_layer.group_send(
            self.room,
            {"type": "chat.message", "message": content["message"]},
        )

    async def chat_message(self, event):
        await self.send_json({"message": event["message"]})

from sqlalchemy import String, Integer, ForeignKey, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    posts: Mapped[list["Post"]] = relationship(back_populates="author", lazy="selectin")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    author: Mapped["User"] = relationship(back_populates="posts")

# 非同期エンジン設定
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    pool_size=20,
    max_overflow=10,
    pool_pre_ping=True,
    echo=False,
)

async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession

class UserRepository:
    def __init__(self, session: AsyncSession):
        self.session = session

    async def get_by_id(self, user_id: int) -> User | None:
        result = await self.session.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()

    async def list_with_posts(self, skip: int = 0, limit: int = 20) -> list[User]:
        result = await self.session.execute(
            select(User)
            .options(selectinload(User.posts))
            .offset(skip)
            .limit(limit)
            .order_by(User.created_at.desc())
        )
        return list(result.scalars().all())

    async def create(self, username: str, email: str) -> User:
        user = User(username=username, email=email)
        self.session.add(user)
        await self.session.commit()
        await self.session.refresh(user)
        return user

    async def update_email(self, user_id: int, new_email: str) -> bool:
        result = await self.session.execute(
            update(User).where(User.id == user_id).values(email=new_email)
        )
        await self.session.commit()
        return result.rowcount > 0

# 初期化
alembic init alembic

# マイグレーション生成
alembic revision --autogenerate -m "add users table"

# マイグレーション適用
alembic upgrade head

# ロールバック
alembic downgrade -1

import asyncio
import aiohttp

# 複数API同時呼び出し
async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return await response.json()

async def fetch_all(urls: list[str]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 実行
urls = [
    "https://api.example.com/users",
    "https://api.example.com/products",
    "https://api.example.com/orders",
]
results = asyncio.run(fetch_all(urls))

import asyncio

# 同時に最大10個のみ実行
semaphore = asyncio.Semaphore(10)

async def rate_limited_fetch(session, url):
    async with semaphore:
        async with session.get(url) as response:
            return await response.json()

非同期が良い場合：
  - 外部API呼び出しが多い場合
  - データベースI/Oが頻繁な場合
  - WebSocket / リアルタイム通信
  - ファイルI/Oが多い場合

同期が良い場合：
  - CPU集約的タスク（画像処理、ML推論）
  - シンプルなCRUD操作
  - レガシーライブラリが非同期をサポートしない場合
  - concurrent.futures.ProcessPoolExecutorの使用が適切な場合

from concurrent.futures import ProcessPoolExecutor
import asyncio

def cpu_heavy_task(data: bytes) -> bytes:
    import hashlib
    return hashlib.sha256(data).digest()

async def process_files(files: list[bytes]) -> list[bytes]:
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor(max_workers=4) as executor:
        tasks = [
            loop.run_in_executor(executor, cpu_heavy_task, f)
            for f in files
        ]
        return await asyncio.gather(*tasks)

# celery_app.py
from celery import Celery

app = Celery(
    "myapp",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Tokyo",
    enable_utc=True,
    task_track_started=True,
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_reject_on_worker_lost=True,
)

from celery import shared_task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_jitter=True,
)
def send_email_task(self, to: str, subject: str, body: str):
    try:
        send_email(to, subject, body)
        logger.info(f"Email sent to {to}")
    except Exception as exc:
        logger.error(f"Email failed: {exc}")
        raise self.retry(exc=exc)

@shared_task(bind=True)
def process_ai_inference(self, model_name: str, input_data: dict):
    self.update_state(state="PROCESSING", meta={"progress": 0})
    model = load_model(model_name)
    self.update_state(state="PROCESSING", meta={"progress": 50})
    result = model.predict(input_data)
    self.update_state(state="PROCESSING", meta={"progress": 100})
    return result

from celery.schedules import crontab

app.conf.beat_schedule = {
    "cleanup-expired-sessions": {
        "task": "tasks.cleanup_sessions",
        "schedule": crontab(hour=2, minute=0),  # 毎日午前2時
    },
    "sync-external-data": {
        "task": "tasks.sync_data",
        "schedule": 300.0,  # 5分ごと
    },
    "weekly-report": {
        "task": "tasks.generate_report",
        "schedule": crontab(hour=9, minute=0, day_of_week="mon"),
    },
}

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser

app = FastAPI()

llm = ChatOpenAI(model="gpt-4o", temperature=0.7, streaming=True)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant. Answer in Japanese."),
    ("human", "{question}"),
])

chain = prompt | llm | StrOutputParser()

@app.post("/chat")
async def chat(question: str):
    response = await chain.ainvoke({"question": question})
    return {"answer": response}

@app.post("/chat/stream")
async def chat_stream(question: str):
    async def generate():
        async for chunk in chain.astream({"question": question}):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "Answer based on the following context:\n{context}"),
    ("human", "{question}"),
])

@app.post("/rag/query")
async def rag_query(question: str):
    docs = await retriever.ainvoke(question)
    context = "\n".join([doc.page_content for doc in docs])

    response = await (rag_prompt | llm | StrOutputParser()).ainvoke(
        {"context": context, "question": question}
    )

    return {
        "answer": response,
        "sources": [{"content": d.page_content, "metadata": d.metadata} for d in docs],
    }

from contextlib import asynccontextmanager
import torch

ml_models = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    ml_models["sentiment"] = load_sentiment_model()
    ml_models["ner"] = load_ner_model()
    yield
    ml_models.clear()
    torch.cuda.empty_cache()

app = FastAPI(lifespan=lifespan)

@app.post("/predict/sentiment")
async def predict_sentiment(text: str):
    model = ml_models["sentiment"]
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, model.predict, text)
    return {"sentiment": result}

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

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

@pytest.mark.anyio
async def test_create_user(client: AsyncClient):
    response = await client.post("/users", json={
        "username": "testuser",
        "email": "test@example.com",
        "age": 25,
    })
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "testuser"

@pytest.mark.anyio
async def test_get_user_not_found(client: AsyncClient):
    response = await client.get("/users/0")
    assert response.status_code == 404

import factory
from factory.alchemy import SQLAlchemyModelFactory

class UserFactory(SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "commit"

    username = factory.Sequence(lambda n: f"user_{n}")
    email = factory.LazyAttribute(lambda o: f"{o.username}@example.com")

class PostFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Post
        sqlalchemy_session_persistence = "commit"

    title = factory.Faker("sentence")
    content = factory.Faker("paragraph")
    author = factory.SubFactory(UserFactory)

# 開発
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# プロダクション — Gunicorn + Uvicorn Workers
gunicorn main:app \
  -w 4 \
  -k uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 120 \
  --graceful-timeout 30 \
  --access-logfile - \
  --error-logfile -

# Stage 1: ビルド
FROM python:3.12-slim AS builder

WORKDIR /app
RUN pip install --no-cache-dir poetry

COPY pyproject.toml poetry.lock ./
RUN poetry export -f requirements.txt --output requirements.txt --without-hashes
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2: ランタイム
FROM python:3.12-slim

WORKDIR /app
COPY --from=builder /install /usr/local
COPY . .

RUN adduser --disabled-password --gecos "" appuser
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD python -c "import httpx; httpx.get('http://localhost:8000/health')"

CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

import structlog
import logging

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    logger_factory=structlog.PrintLoggerFactory(),
)

logger = structlog.get_logger()

async def create_order(order: OrderCreate):
    logger.info("order_created", order_id=order.id, user_id=order.user_id, amount=order.total)

from fastapi import FastAPI
from datetime import datetime

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat(),
        "version": "1.0.0",
    }

@app.get("/health/ready")
async def readiness_check(db: AsyncSession = Depends(get_db)):
    try:
        await db.execute(text("SELECT 1"))
        db_status = "connected"
    except Exception:
        db_status = "disconnected"
        raise HTTPException(status_code=503)

    return {"status": "ready", "database": db_status}