Python 后端开发深度指南：Flask、Django 与 FastAPI 全方位解析

# SQLAlchemy 2.0 声明式模型示例
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Session, relationship
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(80), unique=True, nullable=False)
    email = Column(String(120), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    user_id = Column(Integer, ForeignKey('users.id'))
    created_at = Column(DateTime, default=datetime.utcnow)

    author = relationship('User', back_populates='posts')

# 使用示例
engine = create_engine('sqlite:///blog.db', echo=True)
Base.metadata.create_all(engine)

with Session(engine) as session:
    user = User(username='zhangsan', email='zhangsan@example.com')
    session.add(user)
    session.commit()

django_project/
├── config/                  # Django 项目配置
│   ├── __init__.py
│   ├── settings/
│   │   ├── __init__.py
│   │   ├── base.py         # 公共配置
│   │   ├── development.py  # 开发环境配置
│   │   └── production.py   # 生产环境配置
│   ├── urls.py             # 根路由
│   └── wsgi.py
├── apps/                    # 所有应用集中在 apps 目录
│   ├── users/              # 用户应用
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── views.py
│   │   ├── serializers.py  # DRF 序列化器
│   │   ├── admin.py
│   │   ├── urls.py
│   │   ├── tests/
│   │   └── templates/
│   └── blog/               # 博客应用
│       ├── __init__.py
│       ├── models.py
│       ├── views.py
│       ├── admin.py
│       └── urls.py
├── static/                  # 静态文件
├── media/                   # 媒体文件
├── templates/               # 全局模板
│   └── base.html
├── requirements/
│   ├── base.txt
│   ├── dev.txt
│   └── prod.txt
├── manage.py
├── Dockerfile
└── docker-compose.yml

fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py             # FastAPI 应用入口
│   ├── config.py           # 配置（Pydantic Settings）
│   ├── dependencies.py     # 依赖注入
│   ├── models/             # SQLAlchemy 模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/            # Pydantic 模型（请求/响应）
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── post.py
│   ├── api/                # 路由层
│   │   ├── __init__.py
│   │   ├── v1/            # API 版本控制
│   │   │   ├── __init__.py
│   │   │   ├── users.py
│   │   │   └── posts.py
│   ├── services/           # 业务逻辑
│   │   ├── __init__.py
│   │   └── user_service.py
│   ├── core/               # 基础设施
│   │   ├── __init__.py
│   │   ├── database.py    # 数据库连接
│   │   ├── security.py    # 认证/安全
│   │   └── logging.py     # 日志配置
│   └── tasks/              # 后台任务
│       └── __init__.py
├── tests/
│   ├── conftest.py
│   ├── test_api/
│   └── test_services/
├── alembic/                # 数据库迁移
│   └── versions/
├── alembic.ini
├── .env
├── pyproject.toml
├── Dockerfile
└── docker-compose.yml

# app/models.py - 数据模型
from . import db
from datetime import datetime

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    posts = db.relationship('Post', backref='author', lazy='dynamic')

    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'created_at': self.created_at.isoformat(),
        }

class Post(db.Model):
    __tablename__ = 'posts'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'content': self.content,
            'user_id': self.user_id,
            'created_at': self.created_at.isoformat(),
        }

# app/routes.py - 路由/视图
from flask import Blueprint, request, jsonify
from . import db
from .models import User, Post

api_bp = Blueprint('api', __name__)

# 创建用户
@api_bp.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    if not data or 'username' not in data or 'email' not in data:
        return jsonify({'error': '缺少必填字段'}), 400

    if User.query.filter_by(username=data['username']).first():
        return jsonify({'error': '用户名已存在'}), 409

    user = User(username=data['username'], email=data['email'])
    db.session.add(user)
    db.session.commit()

    return jsonify(user.to_dict()), 201

# 获取用户列表
@api_bp.route('/users', methods=['GET'])
def get_users():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    pagination = User.query.paginate(page=page, per_page=per_page, error_out=False)

    return jsonify({
        'users': [user.to_dict() for user in pagination.items],
        'total': pagination.total,
        'page': pagination.page,
        'pages': pagination.pages,
    })

# 创建文章
@api_bp.route('/posts', methods=['POST'])
def create_post():
    data = request.get_json()
    if not data or 'title' not in data or 'content' not in data or 'user_id' not in data:
        return jsonify({'error': '缺少必填字段'}), 400

    user = User.query.get(data['user_id'])
    if not user:
        return jsonify({'error': '用户不存在'}), 404

    post = Post(
        title=data['title'],
        content=data['content'],
        author=user
    )
    db.session.add(post)
    db.session.commit()

    return jsonify(post.to_dict()), 201

# 获取用户的所有文章
@api_bp.route('/users/<int:user_id>/posts', methods=['GET'])
def get_user_posts(user_id):
    user = User.query.get_or_404(user_id)
    posts = user.posts.order_by(Post.created_at.desc()).all()
    return jsonify([post.to_dict() for post in posts])

# 通用错误处理
@api_bp.errorhandler(404)
def not_found(error):
    return jsonify({'error': '资源未找到'}), 404

@api_bp.errorhandler(500)
def internal_error(error):
    return jsonify({'error': '服务器内部错误'}), 500

# blog/models.py - Django 模型
from django.db import models
from django.contrib.auth.models import AbstractUser
from django.utils.translation import gettext_lazy as _

class User(AbstractUser):
    """自定义用户模型"""
    bio = models.TextField(_('个人简介'), max_length=500, blank=True)
    avatar = models.ImageField(_('头像'), upload_to='avatars/', blank=True)
    website = models.URLField(_('个人网站'), blank=True)
    created_at = models.DateTimeField(_('创建时间'), auto_now_add=True)

    class Meta:
        verbose_name = _('用户')
        verbose_name_plural = _('用户')
        ordering = ['-date_joined']

    def __str__(self):
        return self.username

class Category(models.Model):
    """文章分类"""
    name = models.CharField(_('分类名称'), max_length=100, unique=True)
    slug = models.SlugField(_('URL 别名'), max_length=120, unique=True)
    description = models.TextField(_('描述'), blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        verbose_name = _('分类')
        verbose_name_plural = _('分类')
        ordering = ['name']

    def __str__(self):
        return self.name

class Tag(models.Model):
    """文章标签"""
    name = models.CharField(_('标签'), max_length=50, unique=True)
    slug = models.SlugField(_('URL 别名'), max_length=60, unique=True)

    class Meta:
        verbose_name = _('标签')
        verbose_name_plural = _('标签')

    def __str__(self):
        return self.name

class Post(models.Model):
    """文章模型"""
    STATUS_CHOICES = [
        ('draft', _('草稿')),
        ('published', _('已发布')),
        ('archived', _('已归档')),
    ]

    title = models.CharField(_('标题'), max_length=250)
    slug = models.SlugField(_('URL 别名'), max_length=300, unique=True)
    content = models.TextField(_('正文'))
    excerpt = models.TextField(_('摘要'), max_length=500, blank=True,
                                help_text=_('留空则自动取正文前200字'))
    author = models.ForeignKey(
        User, on_delete=models.CASCADE,
        related_name='posts', verbose_name=_('作者')
    )
    category = models.ForeignKey(
        Category, on_delete=models.SET_NULL,
        null=True, related_name='posts', verbose_name=_('分类')
    )
    tags = models.ManyToManyField(Tag, blank=True, related_name='posts',
                                   verbose_name=_('标签'))
    status = models.CharField(_('状态'), max_length=10,
                               choices=STATUS_CHOICES, default='draft')
    views_count = models.PositiveIntegerField(_('浏览量'), default=0)
    is_pinned = models.BooleanField(_('置顶'), default=False)
    created_at = models.DateTimeField(_('创建时间'), auto_now_add=True)
    updated_at = models.DateTimeField(_('更新时间'), auto_now=True)

    class Meta:
        verbose_name = _('文章')
        verbose_name_plural = _('文章')
        ordering = ['-is_pinned', '-created_at']
        indexes = [
            models.Index(fields=['status', 'created_at']),
            models.Index(fields=['slug']),
        ]

    def __str__(self):
        return self.title

    def save(self, *args, **kwargs):
        if not self.excerpt and self.content:
            # 自动生成摘要
            self.excerpt = self.content[:200]
        super().save(*args, **kwargs)

    def get_absolute_url(self):
        from django.urls import reverse
        return reverse('blog:post_detail', kwargs={'slug': self.slug})

# blog/admin.py - Django Admin 配置
from django.contrib import admin
from django.utils.html import format_html
from django.urls import reverse
from .models import User, Category, Tag, Post

@admin.register(User)
class UserAdmin(admin.ModelAdmin):
    list_display = ('username', 'email', 'is_staff', 'post_count', 'date_joined')
    list_filter = ('is_staff', 'is_active', 'date_joined')
    search_fields = ('username', 'email')
    fieldsets = (
        (None, {'fields': ('username', 'password')}),
        ('个人信息', {'fields': ('email', 'bio', 'avatar', 'website')}),
        ('权限', {'fields': ('is_active', 'is_staff', 'is_superuser', 'groups', 'user_permissions')}),
        ('重要日期', {'fields': ('last_login', 'date_joined')}),
    )

    def post_count(self, obj):
        return obj.posts.count()
    post_count.short_description = '文章数'

@admin.register(Category)
class CategoryAdmin(admin.ModelAdmin):
    list_display = ('name', 'slug', 'post_count', 'created_at')
    prepopulated_fields = {'slug': ('name',)}

    def post_count(self, obj):
        return obj.posts.count()
    post_count.short_description = '文章数'

class PostTagInline(admin.TabularInline):
    model = Post.tags.through
    extra = 1

@admin.register(Post)
class PostAdmin(admin.ModelAdmin):
    list_display = ('title', 'author_link', 'category', 'status', 'views_count',
                    'is_pinned', 'created_at', 'updated_at')
    list_filter = ('status', 'category', 'tags', 'is_pinned', 'created_at')
    search_fields = ('title', 'content', 'excerpt')
    date_hierarchy = 'created_at'
    prepopulated_fields = {'slug': ('title',)}
    list_editable = ('status', 'is_pinned')
    list_select_related = ('author', 'category')
    filter_horizontal = ('tags',)
    actions = ['make_published', 'make_archived']

    fieldsets = (
        (None, {
            'fields': ('title', 'slug', 'author', 'status')
        }),
        ('内容', {
            'fields': ('content', 'excerpt'),
            'classes': ('wide',),
        }),
        ('分类与标签', {
            'fields': ('category', 'tags'),
        }),
        ('统计', {
            'fields': ('views_count', 'is_pinned'),
            'classes': ('collapse',),
        }),
    )

    def author_link(self, obj):
        url = reverse('admin:blog_user_change', args=[obj.author_id])
        return format_html('<a href="{}">{}</a>', url, obj.author.username)
    author_link.short_description = '作者'
    author_link.admin_order_field = 'author'

    @admin.action(description='标记为已发布')
    def make_published(self, request, queryset):
        updated = queryset.update(status='published')
        self.message_user(request, f'已发布 {updated} 篇文章')

    @admin.action(description='标记为已归档')
    def make_archived(self, request, queryset):
        updated = queryset.update(status='archived')
        self.message_user(request, f'已归档 {updated} 篇文章')

# app/main.py - FastAPI 应用入口
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from .core.database import engine, Base
from .api.v1 import users, posts

@asynccontextmanager
async def lifespan(app: FastAPI):
    """管理应用生命周期"""
    # 启动时创建表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # 关闭时释放资源
    await engine.dispose()

app = FastAPI(
    title="Python 后端指南 - FastAPI 示例",
    description="一个展示 FastAPI 异步能力的博客 API",
    version="1.0.0",
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(users.router, prefix="/api/v1/users", tags=["用户"])
app.include_router(posts.router, prefix="/api/v1/posts", tags=["文章"])

# app/api/v1/users.py - 异步路由
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from ...core.database import get_db
from ...models.user import User
from ...schemas.user import UserCreate, UserResponse, UserListResponse
from ...models.post import Post

router = APIRouter()

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(user_data: UserCreate, db: AsyncSession = Depends(get_db)):
    """创建新用户"""
    # 检查用户名是否已存在
    result = await db.execute(
        select(User).where(User.username == user_data.username)
    )
    if result.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="用户名已存在")

    # 检查邮箱是否已存在
    result = await db.execute(
        select(User).where(User.email == user_data.email)
    )
    if result.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="邮箱已被注册")

    user = User(**user_data.model_dump())
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@router.get("/", response_model=UserListResponse)
async def list_users(
    page: int = Query(1, ge=1, description="页码"),
    per_page: int = Query(10, ge=1, le=100, description="每页数量"),
    db: AsyncSession = Depends(get_db),
):
    """分页获取用户列表（含文章数统计）"""
    # 查询总数
    count_result = await db.execute(select(func.count(User.id)))
    total = count_result.scalar()

    # 分页查询
    offset = (page - 1) * per_page
    result = await db.execute(
        select(User).offset(offset).limit(per_page).order_by(User.created_at.desc())
    )
    users = result.scalars().all()

    # 统计每个用户的文章数
    user_list = []
    for user in users:
        post_count = await db.execute(
            select(func.count(Post.id)).where(Post.user_id == user.id)
        )
        user_response = UserResponse(
            id=user.id,
            username=user.username,
            email=user.email,
            created_at=user.created_at,
            post_count=post_count.scalar(),
        )
        user_list.append(user_response)

    return UserListResponse(
        users=user_list,
        total=total,
        page=page,
        pages=(total + per_page - 1) // per_page,
    )

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    """获取单个用户详情"""
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

    post_count = await db.execute(
        select(func.count(Post.id)).where(Post.user_id == user.id)
    )
    return UserResponse(
        id=user.id,
        username=user.username,
        email=user.email,
        created_at=user.created_at,
        post_count=post_count.scalar(),
    )

@router.delete("/{user_id}", status_code=204)
async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)):
    """删除用户及其所有文章"""
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

    await db.delete(user)  # CASCADE 会级联删除文章
    await db.commit()

# app/core/database.py - 异步数据库配置
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from .config import settings

class Base(DeclarativeBase):
    pass

# 异步引擎
engine = create_async_engine(
    settings.DATABASE_URL,  # e.g. "postgresql+asyncpg://user:pass@localhost/db"
    echo=settings.DB_ECHO,
    pool_size=20,
    max_overflow=10,
)

# 异步会话工厂
async_session_factory = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_db() -> AsyncSession:
    """FastAPI 依赖：获取数据库会话"""
    async with async_session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# tests/conftest.py
import pytest
from app import create_app, db

@pytest.fixture
def app():
    app = create_app('testing')
    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.fixture
def runner(app):
    return app.test_cli_runner()

# tests/test_api.py
def test_create_user(client):
    response = client.post('/api/users', json={
        'username': 'testuser',
        'email': 'test@example.com'
    })
    assert response.status_code == 201
    data = response.get_json()
    assert data['username'] == 'testuser'

def test_get_users_pagination(client):
    # 创建测试数据
    for i in range(15):
        client.post('/api/users', json={
            'username': f'user{i}',
            'email': f'user{i}@example.com'
        })

    response = client.get('/api/users?page=1&per_page=10')
    assert response.status_code == 200
    data = response.get_json()
    assert len(data['users']) == 10
    assert data['total'] == 15
    assert data['pages'] == 2

# blog/tests/test_models.py
from django.test import TestCase
from django.contrib.auth import get_user_model
from ..models import Post, Category

User = get_user_model()

class PostModelTest(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username='testuser',
            email='test@example.com',
            password='testpass123'
        )
        self.category = Category.objects.create(
            name='Python',
            slug='python'
        )
        self.post = Post.objects.create(
            title='测试文章',
            slug='test-post',
            content='这是一篇测试文章的内容。',
            author=self.user,
            category=self.category,
            status='published'
        )

    def test_post_creation(self):
        self.assertEqual(self.post.title, '测试文章')
        self.assertEqual(self.post.author.username, 'testuser')
        self.assertTrue(self.post.is_pinned is False)

    def test_post_str(self):
        self.assertEqual(str(self.post), '测试文章')

    def test_auto_excerpt(self):
        """测试自动生成摘要"""
        self.assertEqual(self.post.excerpt, '这是一篇测试文章的内容。')

    def test_post_absolute_url(self):
        url = self.post.get_absolute_url()
        self.assertIn('/blog/test-post/', url)

    def test_post_ordering(self):
        """测试文章默认排序：置顶优先，然后按创建时间降序"""
        post2 = Post.objects.create(
            title='第二篇文章',
            slug='second-post',
            content='内容',
            author=self.user,
            category=self.category,
            is_pinned=True,
        )
        posts = Post.objects.all()
        self.assertEqual(posts[0], post2)  # 置顶的排在前面

# tests/conftest.py
import pytest
import asyncio
from typing import AsyncGenerator
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.main import app
from app.core.database import Base, get_db

# 使用内存 SQLite 进行测试
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"

@pytest.fixture(scope="session")
def event_loop():
    """创建会话级别的事件循环"""
    loop = asyncio.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
    await engine.dispose()

@pytest.fixture
async def test_session(test_engine) -> AsyncGenerator[AsyncSession, None]:
    session_factory = async_sessionmaker(test_engine, expire_on_commit=False)
    async with session_factory() as session:
        yield session

@pytest.fixture
async def client(test_session) -> AsyncGenerator[AsyncClient, None]:
    """创建测试客户端，注入测试数据库会话"""
    async def override_get_db():
        yield test_session

    app.dependency_overrides[get_db] = override_get_db

    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

    app.dependency_overrides.clear()

# tests/test_users_api.py
@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/api/v1/users/", json={
        "username": "testuser",
        "email": "test@example.com"
    })
    assert response.status_code == 201
    data = response.json()
    assert data["username"] == "testuser"
    assert "id" in data

@pytest.mark.asyncio
async def test_create_duplicate_user(client: AsyncClient):
    # 创建第一个用户
    await client.post("/api/v1/users/", json={
        "username": "testuser",
        "email": "test@example.com"
    })
    # 尝试创建同名用户
    response = await client.post("/api/v1/users/", json={
        "username": "testuser",
        "email": "other@example.com"
    })
    assert response.status_code == 409  # Conflict
    assert "已存在" in response.json()["detail"]

# Dockerfile（多阶段构建）
# ========== 构建阶段 ==========
FROM python:3.12-slim AS builder

WORKDIR /build

# 安装编译依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt

# ========== 运行阶段 ==========
FROM python:3.12-slim

RUN groupadd -r appuser && useradd -r -g appuser appuser

# 复制编译好的依赖
COPY --from=builder /root/.local /home/appuser/.local
ENV PATH=/home/appuser/.local/bin:$PATH

WORKDIR /app

# 复制应用代码
COPY --chown=appuser:appuser . .

USER appuser

EXPOSE 8000

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

# 根据实际框架选择启动命令
# Flask: CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "app:create_app()"]
# Django: CMD ["gunicorn", "config.wsgi:application", "-w", "4", "-b", "0.0.0.0:8000"]
# FastAPI: CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
CMD ["gunicorn", "-w", "4", "-b", "0.0.0.0:8000", "app:create_app()"]

# docker-compose.yml
version: '3.8'

services:
  app:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/blog
      - REDIS_URL=redis://redis:6379/0
      - ENVIRONMENT=production
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    restart: unless-stopped
    volumes:
      - static_volume:/app/static
      - media_volume:/app/media
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

  db:
    image: postgres:16-alpine
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
      - POSTGRES_DB=blog
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d blog"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - static_volume:/static:ro
      - media_volume:/media:ro
    depends_on:
      - app
    restart: unless-stopped

volumes:
  postgres_data:
  redis_data:
  static_volume:
  media_volume:

# nginx.conf - 反向代理配置
upstream backend {
    server app:8000;
    keepalive 64;
}

server {
    listen 80;
    server_name api.example.com;

    # 静态文件
    location /static/ {
        alias /static/;
        expires 30d;
        add_header Cache-Control "public, immutable";
    }

    location /media/ {
        alias /media/;
        expires 7d;
        add_header Cache-Control "public";
    }

    # API 反向代理
    location / {
        proxy_pass http://backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # 超时配置
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;

        # 缓冲配置
        proxy_buffering on;
        proxy_buffer_size 8k;
        proxy_buffers 8 32k;
        proxy_busy_buffers_size 64k;

        # 文件上传
        client_max_body_size 50m;
        client_body_buffer_size 128k;
    }

    # 健康检查端点不记录日志
    location /health {
        proxy_pass http://backend;
        access_log off;
    }
}