Python Web開発ガイド：Django vs FastAPI vs Flask、SQLAlchemy、Celery、デプロイ

1. Django vs FastAPI vs Flask — When to Use Which?

Python has three mainstream web frameworks, each suited to different scenarios. Choosing the right framework is the first step to project success — the wrong choice leads to significant technical debt down the line.

Framework Comparison: Django vs FastAPI vs Flask
=================================================

Feature               Django          FastAPI         Flask
-------               ------          -------         -----
Type                  Full-stack      API-first       Micro
ORM                   Built-in        None (SQLAlch.) None (SQLAlch.)
Admin Panel           Built-in        None            None
Auth System           Built-in        Manual/Libs     Manual/Libs
Migrations            Built-in        Alembic         Alembic
Async Support         Partial (3.1+)  Native (ASGI)   No (WSGI)
Data Validation       Forms/DRF       Pydantic        Manual/Marshmallow
Auto API Docs         No              Yes (OpenAPI)   No
Performance           Medium          Very High       Medium
Learning Curve        Steep           Medium          Low
Best For              Full-stack apps APIs/Microsvcs  Prototypes/Simple

Choose Django when:
  - You need a relational database with admin interface
  - Building a CMS, e-commerce, or content platform
  - Team is familiar with Django ecosystem
  - Need batteries-included: auth, sessions, forms

Choose FastAPI when:
  - Building REST APIs or GraphQL backends
  - Need high throughput with async I/O
  - Serving ML models or data science APIs
  - Want automatic Swagger/ReDoc documentation

Choose Flask when:
  - Building a small microservice or prototype
  - Need maximum flexibility in stack assembly
  - Minimal overhead and simplicity are priorities

2. FastAPI Deep Dive: Async Routes, Pydantic Models, and JWT Auth

FastAPI is built on Starlette (ASGI framework) and Pydantic (data validation), delivering near Node.js and Go performance while keeping Python's developer experience. Its automatically generated OpenAPI documentation (Swagger UI and ReDoc) is a standout feature for public APIs and team collaboration.

Project Structure and Basic Setup

# Install dependencies
pip install fastapi uvicorn[standard] pydantic pydantic-settings
pip install sqlalchemy asyncpg alembic python-jose[cryptography] passlib[bcrypt]

# Recommended project structure
myapp/
  app/
    __init__.py
    main.py          # FastAPI app instance, startup/shutdown events
    config.py        # Settings via pydantic-settings
    database.py      # SQLAlchemy engine and session
    models/          # SQLAlchemy ORM models
      __init__.py
      user.py
    schemas/         # Pydantic request/response models
      __init__.py
      user.py
    routers/         # APIRouter modules
      __init__.py
      users.py
      auth.py
    services/        # Business logic
      user_service.py
    dependencies.py  # Shared Depends() functions
  tests/
    conftest.py
    test_users.py
  alembic/
    versions/
  alembic.ini
  requirements.txt
  Dockerfile

Async Routes and Pydantic Models

# app/schemas/user.py
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from datetime import datetime


class UserBase(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_]+$")
    full_name: Optional[str] = Field(None, max_length=100)


class UserCreate(UserBase):
    password: str = Field(..., min_length=8)


class UserUpdate(BaseModel):
    full_name: Optional[str] = None
    email: Optional[EmailStr] = None


class UserResponse(UserBase):
    id: int
    is_active: bool
    created_at: datetime

    model_config = {"from_attributes": True}  # Pydantic v2: replaces orm_mode


# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List

from app.database import get_db
from app.schemas.user import UserCreate, UserResponse, UserUpdate
from app.services.user_service import UserService
from app.dependencies import get_current_user


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


@router.get("/", response_model=List[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 100,
    db: AsyncSession = Depends(get_db),
    current_user = Depends(get_current_user),
):
    """List all users (paginated). Requires authentication."""
    service = UserService(db)
    return await service.get_users(skip=skip, limit=limit)


@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    db: AsyncSession = Depends(get_db),
):
    """Register a new user."""
    service = UserService(db)
    existing = await service.get_by_email(user_in.email)
    if existing:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Email already registered",
        )
    return await service.create_user(user_in)


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
):
    service = UserService(db)
    user = await service.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

JWT Authentication Implementation

# app/routers/auth.py
from datetime import datetime, timedelta
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.database import get_db
from app.services.user_service import UserService


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


def create_access_token(data: dict, expires_delta: timedelta = timedelta(hours=1)):
    payload = data.copy()
    payload["exp"] = datetime.utcnow() + expires_delta
    return jwt.encode(payload, settings.secret_key, algorithm="HS256")


@router.post("/token")
async def login(
    form: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db),
):
    service = UserService(db)
    user = await service.get_by_email(form.username)
    if not user or not pwd_context.verify(form.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    token = create_access_token({"sub": str(user.id)})
    return {"access_token": token, "token_type": "bearer"}


# app/dependencies.py
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    service = UserService(db)
    user = await service.get_by_id(int(user_id))
    if user is None:
        raise credentials_exception
    return user

3. Django Core: ORM, Views, Templates, and REST Framework

Django is famous for its batteries-included philosophy. Its ORM defines data models as Python classes with automatic migration generation. The built-in Admin interface provides a zero-config data management backend, while Django REST Framework (DRF) provides a complete API building toolkit.

Django ORM: Models and Queries

# blog/models.py
from django.db import models
from django.contrib.auth.models import User
from django.utils.text import slugify


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 Post(models.Model):
    STATUS_DRAFT = "draft"
    STATUS_PUBLISHED = "published"
    STATUS_CHOICES = [(STATUS_DRAFT, "Draft"), (STATUS_PUBLISHED, "Published")]

    title = models.CharField(max_length=255)
    slug = models.SlugField(unique=True)
    author = models.ForeignKey(User, on_delete=models.CASCADE, related_name="posts")
    category = models.ForeignKey(
        Category, on_delete=models.SET_NULL, null=True, blank=True, related_name="posts"
    )
    tags = models.ManyToManyField("Tag", blank=True, related_name="posts")
    body = models.TextField()
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=STATUS_DRAFT)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    published_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        ordering = ["-created_at"]
        indexes = [
            models.Index(fields=["status", "-published_at"]),
            models.Index(fields=["author", "status"]),
        ]

    def save(self, *args, **kwargs):
        if not self.slug:
            self.slug = slugify(self.title)
        super().save(*args, **kwargs)


# ORM queries — efficient patterns
# select_related() for ForeignKey (SQL JOIN, avoids N+1)
posts = Post.objects.select_related("author", "category").filter(
    status=Post.STATUS_PUBLISHED
).order_by("-published_at")[:20]

# prefetch_related() for ManyToMany (separate queries, Python-side join)
posts_with_tags = Post.objects.prefetch_related("tags").filter(
    status=Post.STATUS_PUBLISHED
)

# Aggregation
from django.db.models import Count, Avg
stats = Post.objects.values("category__name").annotate(
    post_count=Count("id")
).order_by("-post_count")

# only() — load specific fields (reduces memory)
titles = Post.objects.only("title", "slug", "published_at").filter(
    status=Post.STATUS_PUBLISHED
)

Django REST Framework (DRF)

# blog/serializers.py
from rest_framework import serializers
from .models import Post, Category


class CategorySerializer(serializers.ModelSerializer):
    class Meta:
        model = Category
        fields = ["id", "name", "slug"]


class PostSerializer(serializers.ModelSerializer):
    author_name = serializers.CharField(source="author.get_full_name", read_only=True)
    category = CategorySerializer(read_only=True)
    category_id = serializers.PrimaryKeyRelatedField(
        queryset=Category.objects.all(), source="category", write_only=True
    )

    class Meta:
        model = Post
        fields = [
            "id", "title", "slug", "author_name", "category", "category_id",
            "body", "status", "created_at", "published_at",
        ]
        read_only_fields = ["id", "slug", "created_at"]


# blog/views.py — ViewSet with Router generates full CRUD
from rest_framework import viewsets, permissions, filters
from rest_framework.decorators import action
from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend


class PostViewSet(viewsets.ModelViewSet):
    serializer_class = PostSerializer
    permission_classes = [permissions.IsAuthenticatedOrReadOnly]
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_fields = ["status", "category"]
    search_fields = ["title", "body"]
    ordering_fields = ["created_at", "published_at"]

    def get_queryset(self):
        return Post.objects.select_related("author", "category").prefetch_related("tags")

    def perform_create(self, serializer):
        serializer.save(author=self.request.user)

    @action(detail=False, methods=["get"])
    def published(self, request):
        qs = self.get_queryset().filter(status=Post.STATUS_PUBLISHED)
        serializer = self.get_serializer(qs, many=True)
        return Response(serializer.data)


# blog/urls.py
from rest_framework.routers import DefaultRouter
from .views import PostViewSet

router = DefaultRouter()
router.register(r"posts", PostViewSet, basename="post")
urlpatterns = router.urls
# Generates: GET/POST /posts/, GET/PUT/PATCH/DELETE /posts/{id}/, GET /posts/published/

4. Flask: Lightweight Apps, Blueprints, and SQLAlchemy

Flask is a micro-framework with a minimal core — functionality is added via extensions as needed. Flask Blueprints organize routes into modules, Flask-SQLAlchemy integrates the SQLAlchemy ORM, and Flask-Migrate (based on Alembic) handles database migrations.

# pip install flask flask-sqlalchemy flask-migrate flask-jwt-extended

# app/__init__.py — Application factory pattern
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()


def create_app(config=None):
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = "postgresql://user:pass@localhost/mydb"
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
    app.config["SECRET_KEY"] = "change-me-in-production"

    db.init_app(app)
    migrate.init_app(app, db)

    # Register blueprints
    from app.auth.routes import auth_bp
    from app.api.routes import api_bp
    app.register_blueprint(auth_bp, url_prefix="/auth")
    app.register_blueprint(api_bp, url_prefix="/api/v1")

    return app


# app/api/routes.py — Blueprint
from flask import Blueprint, jsonify, request, abort
from app import db
from app.models import Post

api_bp = Blueprint("api", __name__)


@api_bp.route("/posts", methods=["GET"])
def list_posts():
    page = request.args.get("page", 1, type=int)
    per_page = request.args.get("per_page", 20, type=int)
    pagination = Post.query.order_by(Post.created_at.desc()).paginate(
        page=page, per_page=per_page, error_out=False
    )
    return jsonify({
        "posts": [p.to_dict() for p in pagination.items],
        "total": pagination.total,
        "pages": pagination.pages,
        "current_page": page,
    })


@api_bp.route("/posts/<int:post_id>", methods=["GET"])
def get_post(post_id):
    post = db.get_or_404(Post, post_id)
    return jsonify(post.to_dict())


@api_bp.errorhandler(404)
def not_found(error):
    return jsonify({"error": "Resource not found"}), 404

5. Python Async Programming: asyncio and aiohttp

Python's asyncio module provides an event loop-based async I/O framework. In web development, async is critical for improving concurrency in I/O-bound services (database queries, external API calls, file operations). aiohttp is the most popular async HTTP client/server library.

# asyncio fundamentals
import asyncio
import aiohttp
from typing import List


# Basic coroutine
async def fetch_user(session: aiohttp.ClientSession, user_id: int) -> dict:
    url = f"https://api.example.com/users/{user_id}"
    async with session.get(url) as response:
        response.raise_for_status()
        return await response.json()


# Fetch multiple URLs concurrently (not sequentially)
async def fetch_all_users(user_ids: List[int]) -> List[dict]:
    async with aiohttp.ClientSession() as session:
        # asyncio.gather runs coroutines concurrently
        tasks = [fetch_user(session, uid) for uid in user_ids]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in results if not isinstance(r, Exception)]


# Timeout and retry with asyncio
async def fetch_with_timeout(url: str, timeout_sec: float = 5.0) -> dict:
    timeout = aiohttp.ClientTimeout(total=timeout_sec)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            async with session.get(url) as resp:
                return await resp.json()
        except asyncio.TimeoutError:
            raise TimeoutError(f"Request to {url} timed out after {timeout_sec}s")


# Run CPU-bound work without blocking the event loop
import concurrent.futures


async def process_image_async(image_data: bytes) -> bytes:
    loop = asyncio.get_event_loop()
    # Run CPU-bound work in a thread pool
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, _process_image_sync, image_data)
    return result


def _process_image_sync(image_data: bytes) -> bytes:
    # Synchronous CPU-bound image processing
    from PIL import Image
    import io
    img = Image.open(io.BytesIO(image_data))
    img = img.resize((800, 600))
    output = io.BytesIO()
    img.save(output, format="JPEG", quality=85)
    return output.getvalue()


# asyncio.Semaphore — limit concurrent connections
async def fetch_with_semaphore(urls: List[str], max_concurrent: int = 10):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_one(session, url):
        async with semaphore:
            async with session.get(url) as resp:
                return await resp.text()

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks)

6. SQLAlchemy 2.0: Core, ORM, and Async Sessions

SQLAlchemy 2.0 brings major updates: a unified 2.0 query style (replacing Session.query() with select() statements), native AsyncSession support, and improved type annotations. SQLAlchemy 2.0 is the preferred ORM for non-Django projects.

ORM Model Definition (2.0 Style)

# app/models/user.py — SQLAlchemy 2.0 declarative style
from sqlalchemy import String, Boolean, DateTime, ForeignKey, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from datetime import datetime
from typing import Optional, List


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    # Relationship: one user has many posts
    posts: Mapped[List["Post"]] = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(255))
    body: Mapped[str] = mapped_column(Text)
    published: Mapped[bool] = mapped_column(Boolean, default=False)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    author: Mapped["User"] = relationship("User", back_populates="posts")

Async Sessions and CRUD Operations

# app/database.py — Async engine and session
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.models import Base

# Use asyncpg driver for PostgreSQL async
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/mydb"

engine = create_async_engine(DATABASE_URL, echo=False, pool_size=10, max_overflow=20)
AsyncSessionLocal = async_sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)


async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


# app/services/user_service.py — SQLAlchemy 2.0 query style
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update, delete
from app.models.user import User
from app.schemas.user import UserCreate
from passlib.context import CryptContext

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


class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get_by_id(self, user_id: int) -> User | None:
        # SQLAlchemy 2.0 style: use select() instead of session.query()
        result = await self.db.execute(select(User).where(User.id == user_id))
        return result.scalar_one_or_none()

    async def get_by_email(self, email: str) -> User | None:
        result = await self.db.execute(select(User).where(User.email == email))
        return result.scalar_one_or_none()

    async def get_users(self, skip: int = 0, limit: int = 100) -> list[User]:
        result = await self.db.execute(
            select(User).where(User.is_active == True).offset(skip).limit(limit)
        )
        return list(result.scalars().all())

    async def create_user(self, user_in: UserCreate) -> User:
        user = User(
            email=user_in.email,
            username=user_in.username,
            hashed_password=pwd_context.hash(user_in.password),
        )
        self.db.add(user)
        await self.db.flush()  # Get the generated id without committing
        await self.db.refresh(user)
        return user

    async def deactivate_user(self, user_id: int) -> None:
        await self.db.execute(
            update(User).where(User.id == user_id).values(is_active=False)
        )

7. Alembic Database Migrations

Alembic is the official migration tool for SQLAlchemy, supporting auto-generation of migration scripts from model changes as well as manual migration logic for complex scenarios. All migration scripts must be committed to version control.

# Initialize Alembic in your project
pip install alembic
alembic init alembic

# alembic/env.py — configure for async SQLAlchemy
import asyncio
from logging.config import fileConfig
from sqlalchemy.ext.asyncio import create_async_engine
from alembic import context
from app.models import Base  # import your models

config = context.config
target_metadata = Base.metadata


def run_migrations_offline():
    context.configure(
        url=config.get_main_option("sqlalchemy.url"),
        target_metadata=target_metadata,
        literal_binds=True,
    )
    with context.begin_transaction():
        context.run_migrations()


async def run_migrations_online():
    connectable = create_async_engine(
        config.get_main_option("sqlalchemy.url")
    )
    async with connectable.connect() as connection:
        await connection.run_sync(
            lambda sync_conn: context.configure(
                connection=sync_conn,
                target_metadata=target_metadata,
            )
        )
        async with connection.begin():
            await connection.run_sync(lambda _: context.run_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    asyncio.run(run_migrations_online())


# Common Alembic commands
# Generate migration from model changes (auto-detect)
alembic revision --autogenerate -m "add users table"

# Apply all pending migrations
alembic upgrade head

# Roll back the last migration
alembic downgrade -1

# Roll back to a specific revision
alembic downgrade abc123

# View migration history
alembic history --verbose

# View current revision
alembic current

8. Celery for Background Task Processing

Celery is the most popular distributed task queue in the Python ecosystem, supporting async tasks, scheduled tasks (Celery Beat), retries, result storage, and task chains. Redis or RabbitMQ are common message brokers, with Redis also recommended as the result backend.

# pip install celery redis

# app/celery_app.py
from celery import Celery

celery_app = Celery(
    "myapp",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
    include=["app.tasks"],
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="UTC",
    enable_utc=True,
    task_acks_late=True,           # Ack only after task completes
    worker_prefetch_multiplier=1,   # Prevent worker overload
)


# app/tasks.py
from app.celery_app import celery_app
import time


@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def send_welcome_email(self, user_id: int, email: str):
    """Send welcome email — retries up to 3 times on failure."""
    try:
        # Your email sending logic here
        _send_email(to=email, subject="Welcome!", body="...")
    except Exception as exc:
        raise self.retry(exc=exc)


@celery_app.task
def process_video(video_id: int, output_format: str = "mp4"):
    """Long-running video processing task."""
    # Process video...
    return {"video_id": video_id, "status": "completed"}


# Call tasks from FastAPI/Django/Flask
# .delay() is a shortcut for .apply_async()
send_welcome_email.delay(user_id=42, email="user@example.com")

# .apply_async() with options
process_video.apply_async(
    args=[video_id],
    kwargs={"output_format": "webm"},
    countdown=10,         # Start after 10 seconds
    expires=3600,         # Discard if not consumed in 1 hour
    priority=5,           # 0-9, higher = higher priority
)


# Celery Beat — Periodic tasks (cron-like scheduling)
from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    "daily-report": {
        "task": "app.tasks.generate_daily_report",
        "schedule": crontab(hour=8, minute=0),  # Every day at 8:00 AM
    },
    "cleanup-sessions": {
        "task": "app.tasks.cleanup_expired_sessions",
        "schedule": 300.0,  # Every 300 seconds
    },
}


# Start workers
celery -A app.celery_app worker --loglevel=info --concurrency=4
celery -A app.celery_app beat --loglevel=info

9. Testing Python Web Apps: pytest and FastAPI TestClient

pytest is the de facto standard for Python testing. FastAPI provides a TestClient based on httpx for testing API endpoints without starting a server. Testing strategy includes unit tests (business logic), integration tests (database interaction), and end-to-end tests (full API flows).

# pip install pytest pytest-asyncio httpx pytest-cov

# tests/conftest.py — shared fixtures
import pytest
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.main import app
from app.database import get_db
from app.models import Base

TEST_DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/test_db"


@pytest.fixture(scope="session")
def event_loop():
    import asyncio
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="session")
async def test_engine():
    engine = create_async_engine(TEST_DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await engine.dispose()


@pytest.fixture
async def db_session(test_engine):
    TestSession = async_sessionmaker(test_engine, expire_on_commit=False)
    async with TestSession() as session:
        yield session
        await session.rollback()  # Roll back after each test


@pytest.fixture
async def client(db_session):
    def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac
    app.dependency_overrides.clear()


# tests/test_users.py
import pytest


@pytest.mark.asyncio
async def test_create_user(client):
    response = await client.post("/users/", json={
        "email": "test@example.com",
        "username": "testuser",
        "password": "securepassword123",
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "test@example.com"
    assert "id" in data
    assert "password" not in data  # Never expose passwords


@pytest.mark.asyncio
async def test_duplicate_email_returns_409(client):
    payload = {"email": "dup@example.com", "username": "user1", "password": "pass12345"}
    await client.post("/users/", json=payload)
    # Second registration with same email
    response = await client.post("/users/", json={**payload, "username": "user2"})
    assert response.status_code == 409


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


# Run tests
pytest tests/ -v --cov=app --cov-report=html

10. Production Deployment: Gunicorn, Uvicorn, and Docker

For production deployment of Python web apps, use Gunicorn as the process manager with UvicornWorker (for ASGI/FastAPI) or the default sync worker (for Django/Flask). Always run Python services behind a reverse proxy like Nginx or Caddy to handle SSL termination, static files, and request buffering.

Gunicorn + Uvicorn (FastAPI)

# FastAPI / ASGI app
# -w: worker count (recommended: 2 * CPU cores + 1)
# -k: worker class
# -b: bind address
# --timeout: worker timeout in seconds
gunicorn app.main:app \
    -w 4 \
    -k uvicorn.workers.UvicornWorker \
    -b 0.0.0.0:8000 \
    --timeout 120 \
    --keep-alive 5 \
    --access-logfile - \
    --error-logfile -

# Django / WSGI app
gunicorn myproject.wsgi:application \
    -w 4 \
    -b 0.0.0.0:8000 \
    --timeout 120

# gunicorn.conf.py — configuration file
bind = "0.0.0.0:8000"
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
timeout = 120
keepalive = 5
max_requests = 1000          # Restart workers after N requests (prevent memory leaks)
max_requests_jitter = 100    # Random jitter to prevent all workers restarting at once
loglevel = "info"
accesslog = "-"
errorlog = "-"
preload_app = True           # Load app code before forking workers (saves memory)

Dockerfile for Python Web Apps

# Dockerfile — multi-stage for FastAPI
FROM python:3.12-slim AS builder

WORKDIR /app

# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt


# Production stage
FROM python:3.12-slim

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends libpq5 \
    && rm -rf /var/lib/apt/lists/*

# Copy installed packages from builder
COPY --from=builder /root/.local /root/.local

# Copy application code
COPY . .

# Create non-root user for security
RUN useradd --no-create-home --shell /bin/false appuser
USER appuser

ENV PATH=/root/.local/bin:$PATH
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1

EXPOSE 8000

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


# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:password@db/mydb
      - SECRET_KEY=${SECRET_KEY}
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started

  db:
    image: postgres:16-alpine
    volumes:
      - pgdata:/var/lib/postgresql/data
    environment:
      POSTGRES_PASSWORD: password
      POSTGRES_DB: mydb
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 5s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine

  worker:
    build: .
    command: celery -A app.celery_app worker -l info -c 4
    environment:
      - DATABASE_URL=postgresql+asyncpg://postgres:password@db/mydb
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - db
      - redis

volumes:
  pgdata:

11. Detailed Framework Comparison Table

Here is a detailed comparison of Django, FastAPI, and Flask across multiple dimensions to help you make the final decision for your project:

Frequently Asked Questions

Conclusion

The Python web development ecosystem is mature and robust in 2026. FastAPI has become the preferred framework for new API services, thanks to native async support, automatic OpenAPI documentation, and Pydantic validation. Django maintains its dominant position in full-stack web application development with its batteries-included philosophy. Flask remains relevant for lightweight scenarios requiring maximum flexibility.

Regardless of which framework you choose, certain practices are universal: use SQLAlchemy 2.0 (or Django ORM) for the data layer, Alembic (or Django migrations) for schema changes, Celery for long-running background tasks, pytest for code quality, and Docker + Gunicorn/Uvicorn + Nginx for production deployment. Manage all secrets via environment variables — never hardcode passwords or API keys in your codebase.

Master these tools and patterns and you will be equipped to build reliable, high-performance, maintainable Python web applications — from small prototypes to production systems handling tens of thousands of requests per second.