【Python Web 开发精通】第23讲 | 生产环境部署：让你的应用真正上线

所谓伊人，在水一方333

313人浏览 · 2026-03-30 08:12:37

所谓伊人，在水一方333 · 2026-03-30 08:12:37 发布

环境声明

Python 版本：Python 3.12+ (建议使用 3.10 以上版本)
FastAPI 版本：FastAPI 0.115+
Uvicorn 版本：Uvicorn 0.30+
Gunicorn 版本：Gunicorn 23.0+
Nginx 版本：Nginx 1.24+
开发工具：PyCharm 或 VS Code
操作系统：Linux (Ubuntu 22.04+) 推荐用于生产环境

学习目标

学完本讲，你将能够：

理解生产环境与开发环境的本质区别
掌握 Gunicorn 和 Uvicorn 的配置与优化
配置 Nginx 作为反向代理和负载均衡器
申请和配置 HTTPS/SSL 证书
使用环境变量管理不同环境的配置
完成一个完整的生产环境部署流程

1. 生产环境与开发环境的区别

1.1 核心差异对比

维度	开发环境	生产环境
目标	快速迭代、调试方便	稳定运行、高性能、高可用
调试信息	详细错误堆栈	友好的错误提示，隐藏敏感信息
性能优化	关闭或最小化	开启所有优化（缓存、压缩等）
安全配置	宽松，便于测试	严格，HTTPS、CORS、CSRF 等
日志级别	DEBUG	INFO 或 WARNING
数据库	SQLite 或本地开发库	生产级数据库（PostgreSQL/MySQL）
静态文件	由开发服务器提供	由 Nginx/CDN 提供

1.2 生产环境检查清单

# config/production_checklist.py
"""
生产环境部署检查清单
"""

PRODUCTION_CHECKLIST = {
    "安全性": [
        "[ ] 关闭 DEBUG 模式",
        "[ ] 配置强密码和密钥",
        "[ ] 启用 HTTPS (SSL/TLS)",
        "[ ] 配置安全的 Cookie 属性",
        "[ ] 设置适当的 CORS 策略",
        "[ ] 添加安全响应头",
        "[ ] 验证所有用户输入",
    ],
    "性能": [
        "[ ] 启用 Gzip 压缩",
        "[ ] 配置静态文件缓存",
        "[ ] 使用连接池",
        "[ ] 启用数据库查询缓存",
        "[ ] 配置 CDN 加速",
    ],
    "监控": [
        "[ ] 配置应用日志",
        "[ ] 设置错误告警",
        "[ ] 监控服务器资源",
        "[ ] 配置健康检查端点",
    ],
    "部署": [
        "[ ] 使用进程管理器 (systemd/supervisor)",
        "[ ] 配置自动重启",
        "[ ] 设置环境变量",
        "[ ] 准备回滚方案",
    ]
}

2. ASGI 服务器配置

2.1 Uvicorn 基础配置

# uvicorn_config.py
"""
Uvicorn 生产环境配置文件
"""

import multiprocessing
import os

# 获取 CPU 核心数
workers = multiprocessing.cpu_count() * 2 + 1

# Uvicorn 配置
UVICORN_CONFIG = {
    # 绑定地址和端口
    "host": "0.0.0.0",
    "port": 8000,
    
    # 工作进程配置
    "workers": workers,  # 工作进程数
    "worker_class": "uvicorn.workers.UvicornWorker",
    
    # 连接配置
    "backlog": 2048,  # 等待连接队列大小
    "limit_concurrency": 1000,  # 最大并发连接数
    "limit_max_requests": 10000,  # 每个工作进程最大请求数
    "limit_max_requests_jitter": 100,  # 随机抖动，避免同时重启
    
    # 超时配置
    "timeout_keep_alive": 5,  # 保持连接超时
    "timeout_notify": 30,  # 通知超时
    
    # 日志配置
    "log_level": "info",
    "access_log": True,
    "error_log": True,
    
    # 性能配置
    "reload": False,  # 生产环境禁用自动重载
    "proxy_headers": True,  # 信任代理头
    "forwarded_allow_ips": "*",  # 允许的代理 IP
}

# 命令行启动方式：
# uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

2.2 Gunicorn + Uvicorn 组合配置

# gunicorn.conf.py
"""
Gunicorn 配置文件
Gunicorn 作为进程管理器，UvicornWorker 处理 ASGI 请求
"""

import multiprocessing
import os

# 服务器套接字绑定
bind = "0.0.0.0:8000"

# 工作进程配置
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000

# 工作进程生命周期
max_requests = 10000
max_requests_jitter = 100
timeout = 120
graceful_timeout = 30
keepalive = 5

# 预加载应用（节省内存）
preload_app = True

# 进程名称
proc_name = "myapp"

# 日志配置
accesslog = "/var/log/gunicorn/access.log"
errorlog = "/var/log/gunicorn/error.log"
loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'

# 进程管理
pidfile = "/var/run/gunicorn.pid"
daemon = False  # 生产环境使用 systemd 管理，不启用守护模式

# 安全配置
limit_request_line = 4094
limit_request_fields = 100
limit_request_field_size = 8190

# 钩子函数
def on_starting(server):
    """服务器启动前调用"""
    print("Gunicorn 正在启动...")

def on_reload(server):
    """重新加载配置时调用"""
    print("Gunicorn 正在重新加载...")

def when_ready(server):
    """服务器启动完成后调用"""
    print(f"Gunicorn 已就绪，工作进程数: {workers}")

def worker_int(worker):
    """工作进程收到 SIGINT 或 SIGQUIT 时调用"""
    print(f"工作进程 {worker.pid} 正在退出...")

def on_exit(server):
    """服务器退出时调用"""
    print("Gunicorn 已退出")

2.3 启动脚本

#!/bin/bash
# start_production.sh
# 生产环境启动脚本

# 设置环境变量
export APP_ENV=production
export PYTHONPATH=/opt/myapp
export LOG_LEVEL=info

# 创建日志目录
mkdir -p /var/log/gunicorn
mkdir -p /var/log/myapp

# 启动 Gunicorn
cd /opt/myapp

exec gunicorn \
    -c gunicorn.conf.py \
    -k uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --workers 4 \
    --worker-connections 1000 \
    --max-requests 10000 \
    --max-requests-jitter 100 \
    --timeout 120 \
    --graceful-timeout 30 \
    --keep-alive 5 \
    --access-logfile /var/log/gunicorn/access.log \
    --error-logfile /var/log/gunicorn/error.log \
    --log-level info \
    --preload \
    main:app

2.4 性能调优指南

# performance_tuning.py
"""
ASGI 服务器性能调优指南
"""

import multiprocessing
import psutil


def calculate_optimal_workers():
    """
    计算最优工作进程数
    
    公式：workers = (2 * CPU核心数) + 1
    原理：
    - CPU 密集型：workers = CPU核心数
    - I/O 密集型：workers = (2 * CPU核心数) + 1
    - Web 应用通常是 I/O 密集型
    """
    cpu_count = multiprocessing.cpu_count()
    memory = psutil.virtual_memory()
    
    # 基础计算
    base_workers = (2 * cpu_count) + 1
    
    # 根据内存调整（假设每个工作进程占用 200MB）
    memory_based_workers = memory.available // (200 * 1024 * 1024)
    
    # 取较小值
    optimal_workers = min(base_workers, memory_based_workers)
    
    return max(2, optimal_workers)  # 至少 2 个


def get_uvicorn_settings():
    """获取 Uvicorn 优化配置"""
    cpu_count = multiprocessing.cpu_count()
    
    return {
        # 工作进程
        "workers": (2 * cpu_count) + 1,
        
        # 连接配置
        "backlog": min(2048, 65535),  # 系统限制通常是 128-65535
        "limit_concurrency": 1000,  # 根据数据库连接池调整
        
        # 生命周期管理
        "limit_max_requests": 10000,  # 防止内存泄漏
        "limit_max_requests_jitter": 100,  # 避免同时重启
        
        # 超时配置
        "timeout_keep_alive": 5,  # 保持连接时间
        "timeout_notify": 30,
        
        # HTTP 协议
        "http": "auto",  # 自动选择 h11 或 httptools
        "loop": "auto",  # 自动选择 asyncio 或 uvloop
        
        # 性能优化
        "proxy_headers": True,
        "forwarded_allow_ips": "*",
    }


# 不同场景的配置建议
DEPLOYMENT_SCENARIOS = {
    "小型应用": {
        "workers": 2,
        "worker_connections": 100,
        "description": "个人项目、小型内部工具"
    },
    "中型应用": {
        "workers": 4,
        "worker_connections": 500,
        "description": "中小型 Web 应用、API 服务"
    },
    "大型应用": {
        "workers": 8,
        "worker_connections": 1000,
        "description": "高并发 Web 应用、微服务"
    },
    "超大型应用": {
        "workers": 16,
        "worker_connections": 2000,
        "description": "需要配合负载均衡使用"
    }
}

3. Nginx 反向代理配置

3.1 基础反向代理配置

# /etc/nginx/sites-available/myapp
# Nginx 基础反向代理配置

upstream app_server {
    # 应用服务器地址
    server 127.0.0.1:8000 weight=5;
    # 可以添加更多实例实现负载均衡
    # server 127.0.0.1:8001 weight=5;
    # server 127.0.0.1:8002 backup;  # 备用服务器
    
    keepalive 32;  # 保持连接数
}

server {
    listen 80;
    server_name example.com www.example.com;
    
    # 字符编码
    charset utf-8;
    
    # 客户端请求体大小限制
    client_max_body_size 20M;
    
    # 日志配置
    access_log /var/log/nginx/myapp_access.log;
    error_log /var/log/nginx/myapp_error.log;
    
    # 静态文件服务（Nginx 直接处理，不转发到应用）
    location /static/ {
        alias /opt/myapp/static/;
        expires 30d;
        add_header Cache-Control "public, immutable";
        
        # 开启 Gzip
        gzip on;
        gzip_types text/css application/javascript;
    }
    
    # 媒体文件服务
    location /media/ {
        alias /opt/myapp/media/;
        expires 7d;
        add_header Cache-Control "public";
    }
    
    # 健康检查端点（用于负载均衡器检测）
    location /health {
        access_log off;
        return 200 "healthy\n";
        add_header Content-Type text/plain;
    }
    
    # 反向代理到应用服务器
    location / {
        proxy_pass http://app_server;
        proxy_http_version 1.1;
        
        # 代理头设置
        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_set_header X-Forwarded-Host $host;
        proxy_set_header X-Forwarded-Port $server_port;
        
        # WebSocket 支持（如果需要）
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        
        # 超时设置
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
        
        # 缓冲区设置
        proxy_buffering on;
        proxy_buffer_size 4k;
        proxy_buffers 8 4k;
        proxy_busy_buffers_size 8k;
        
        # 错误处理
        proxy_intercept_errors on;
        error_page 500 502 503 504 /50x.html;
    }
    
    # 错误页面
    location = /50x.html {
        root /usr/share/nginx/html;
        internal;
    }
}

3.2 高级负载均衡配置

# /etc/nginx/sites-available/myapp_loadbalancer
# Nginx 负载均衡配置

upstream backend {
    # 轮询（默认）
    server 127.0.0.1:8000 weight=5;
    server 127.0.0.1:8001 weight=5;
    server 127.0.0.1:8002 weight=3;
    
    # IP 哈希（保持会话）
    # ip_hash;
    
    # 最少连接
    # least_conn;
    
    # 一致性哈希（基于请求 URI）
    # hash $request_uri consistent;
    
    keepalive 64;
    keepalive_timeout 60s;
    keepalive_requests 1000;
}

# 限流区域定义
limit_req_zone $binary_remote_addr zone=api_limit:10m rate=10r/s;
limit_req_zone $binary_remote_addr zone=login_limit:10m rate=1r/s;

# 连接数限制
limit_conn_zone $binary_remote_addr zone=addr:10m;

server {
    listen 80;
    server_name api.example.com;
    
    # 全局安全头
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header X-XSS-Protection "1; mode=block" always;
    add_header Referrer-Policy "strict-origin-when-cross-origin" always;
    
    # API 限流
    location /api/ {
        limit_req zone=api_limit burst=20 nodelay;
        limit_conn addr 10;
        
        proxy_pass http://backend;
        proxy_http_version 1.1;
        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;
        
        # 启用压缩
        gzip on;
        gzip_types application/json;
        gzip_min_length 1000;
    }
    
    # 登录接口特殊限流
    location /api/auth/login {
        limit_req zone=login_limit burst=5 nodelay;
        
        proxy_pass http://backend;
        proxy_http_version 1.1;
        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;
    }
}

3.3 性能优化配置

# /etc/nginx/nginx.conf
# Nginx 主配置文件优化

user nginx;
worker_processes auto;  # 自动根据 CPU 核心数设置
error_log /var/log/nginx/error.log warn;
pid /var/run/nginx.pid;

# 事件模块配置
events {
    worker_connections 4096;  # 每个工作进程的最大连接数
    use epoll;  # Linux 高性能网络 I/O 模型
    multi_accept on;  # 一次接受所有新连接
}

http {
    include /etc/nginx/mime.types;
    default_type application/octet-stream;
    
    # 日志格式
    log_format main '$remote_addr - $remote_user [$time_local] "$request" '
                    '$status $body_bytes_sent "$http_referer" '
                    '"$http_user_agent" "$http_x_forwarded_for" '
                    '$request_time $upstream_response_time';
    
    access_log /var/log/nginx/access.log main;
    
    # 性能优化
    sendfile on;  # 启用 sendfile
    tcp_nopush on;  # 防止网络阻塞
    tcp_nodelay on;  # 实时发送数据
    keepalive_timeout 65;
    types_hash_max_size 2048;
    
    # Gzip 压缩
    gzip on;
    gzip_vary on;
    gzip_proxied any;
    gzip_comp_level 6;
    gzip_types
        text/plain
        text/css
        text/xml
        text/javascript
        application/json
        application/javascript
        application/xml+rss
        application/rss+xml
        font/truetype
        font/opentype
        application/vnd.ms-fontobject
        image/svg+xml;
    gzip_min_length 1000;  # 小于 1KB 不压缩
    
    # 静态文件缓存
    open_file_cache max=1000 inactive=20s;
    open_file_cache_valid 30s;
    open_file_cache_min_uses 2;
    open_file_cache_errors on;
    
    # SSL 配置（稍后详解）
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_prefer_server_ciphers on;
    ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
    ssl_session_cache shared:SSL:10m;
    ssl_session_timeout 10m;
    
    # 包含其他配置
    include /etc/nginx/conf.d/*.conf;
    include /etc/nginx/sites-enabled/*;
}

4. HTTPS 与 SSL 证书配置

4.1 使用 Let’s Encrypt 免费证书

#!/bin/bash
# setup_ssl.sh
# Let's Encrypt SSL 证书自动配置脚本

DOMAIN="example.com"
EMAIL="admin@example.com"
WEBROOT="/var/www/certbot"

# 安装 Certbot
sudo apt update
sudo apt install -y certbot python3-certbot-nginx

# 创建 webroot 目录
sudo mkdir -p $WEBROOT

# 获取证书（使用 webroot 验证）
sudo certbot certonly \
    --webroot \
    --webroot-path=$WEBROOT \
    --domain $DOMAIN \
    --domain www.$DOMAIN \
    --email $EMAIL \
    --agree-tos \
    --non-interactive

# 设置自动续期
# Certbot 会自动添加 systemd timer，也可以手动测试：
# sudo certbot renew --dry-run

echo "SSL 证书已配置完成"
echo "证书位置: /etc/letsencrypt/live/$DOMAIN/"

4.2 HTTPS 服务器配置

# /etc/nginx/sites-available/myapp_ssl
# HTTPS 配置

server {
    listen 80;
    server_name example.com www.example.com;
    
    # HTTP 重定向到 HTTPS
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name example.com www.example.com;
    
    # SSL 证书配置
    ssl_certificate /etc/letsencrypt/live/example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/example.com/privkey.pem;
    
    # SSL 优化配置
    ssl_protocols TLSv1.2 TLSv1.3;
    ssl_prefer_server_ciphers on;
    ssl_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
    ssl_session_cache shared:SSL:50m;
    ssl_session_timeout 1d;
    ssl_session_tickets off;
    
    # OCSP Stapling
    ssl_stapling on;
    ssl_stapling_verify on;
    ssl_trusted_certificate /etc/letsencrypt/live/example.com/chain.pem;
    resolver 8.8.8.8 8.8.4.4 valid=300s;
    resolver_timeout 5s;
    
    # 安全头
    add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload" always;
    add_header X-Frame-Options "SAMEORIGIN" always;
    add_header X-Content-Type-Options "nosniff" always;
    add_header X-XSS-Protection "1; mode=block" always;
    add_header Referrer-Policy "strict-origin-when-cross-origin" always;
    add_header Permissions-Policy "geolocation=(), microphone=(), camera=()" always;
    
    # 字符编码
    charset utf-8;
    
    # 客户端请求体大小限制
    client_max_body_size 20M;
    
    # 日志配置
    access_log /var/log/nginx/myapp_access.log;
    error_log /var/log/nginx/myapp_error.log;
    
    # 静态文件服务
    location /static/ {
        alias /opt/myapp/static/;
        expires 6M;
        access_log off;
        add_header Cache-Control "public, immutable";
        
        # 开启 Gzip
        gzip on;
        gzip_types text/css application/javascript;
    }
    
    # 媒体文件服务
    location /media/ {
        alias /opt/myapp/media/;
        expires 1M;
        access_log off;
        add_header Cache-Control "public";
    }
    
    # 健康检查
    location /health {
        access_log off;
        return 200 "healthy\n";
        add_header Content-Type text/plain;
    }
    
    # 反向代理
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        
        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;
        
        # WebSocket 支持
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        
        # 超时设置
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
    }
}

4.3 证书自动续期

#!/bin/bash
# renew_ssl.sh
# SSL 证书续期脚本

# 测试续期（不实际执行）
certbot renew --dry-run

# 实际续期
certbot renew --quiet

# 续期后重载 Nginx
nginx -t && systemctl reload nginx

# 添加到 crontab（每天凌晨 3 点检查续期）
# 0 3 * * * /opt/myapp/scripts/renew_ssl.sh >> /var/log/certbot-renew.log 2>&1

# 设置自动续期 cron 任务
(crontab -l 2>/dev/null; echo "0 3 * * * /usr/bin/certbot renew --quiet --deploy-hook 'systemctl reload nginx'") | crontab -

4.4 SSL 配置检测工具

# ssl_checker.py
"""
SSL 配置检测工具
"""

import ssl
import socket
import subprocess
from dataclasses import dataclass
from typing import Optional


@dataclass
class SSLInfo:
    """SSL 证书信息"""
    domain: str
    issuer: str
    subject: str
    not_before: str
    not_after: str
    serial_number: str
    version: int
    cipher: str
    protocol: str
    
    @property
    def is_valid(self) -> bool:
        """检查证书是否有效"""
        from datetime import datetime
        not_after = datetime.strptime(self.not_after, '%b %d %H:%M:%S %Y %Z')
        return datetime.now() < not_after


def check_ssl_certificate(domain: str, port: int = 443) -> Optional[SSLInfo]:
    """检查 SSL 证书信息"""
    try:
        context = ssl.create_default_context()
        with socket.create_connection((domain, port), timeout=10) as sock:
            with context.wrap_socket(sock, server_hostname=domain) as ssock:
                cert = ssock.getpeercert()
                cipher = ssock.cipher()
                version = ssock.version()
                
                return SSLInfo(
                    domain=domain,
                    issuer=cert.get('issuer', [('', '')])[0][0][1],
                    subject=cert.get('subject', [('', '')])[0][0][1],
                    not_before=cert.get('notBefore', ''),
                    not_after=cert.get('notAfter', ''),
                    serial_number=cert.get('serialNumber', ''),
                    version=ssock.version(),
                    cipher=cipher[0] if cipher else '',
                    protocol=version
                )
    except Exception as e:
        print(f"检查失败: {e}")
        return None


def check_ssl_rating(domain: str) -> dict:
    """使用 ssllabs-scan 检查 SSL 评级"""
    try:
        result = subprocess.run(
            ['ssllabs-scan', domain],
            capture_output=True,
            text=True,
            timeout=300
        )
        return {
            "success": result.returncode == 0,
            "output": result.stdout,
            "error": result.stderr
        }
    except FileNotFoundError:
        return {
            "success": False,
            "error": "ssllabs-scan 未安装，请访问 https://www.ssllabs.com/ssltest/ 进行在线测试"
        }
    except subprocess.TimeoutExpired:
        return {
            "success": False,
            "error": "检查超时"
        }


# 使用示例
if __name__ == "__main__":
    domain = "example.com"
    info = check_ssl_certificate(domain)
    
    if info:
        print(f"域名: {info.domain}")
        print(f"颁发者: {info.issuer}")
        print(f"有效期至: {info.not_after}")
        print(f"是否有效: {info.is_valid}")
        print(f"协议版本: {info.protocol}")
        print(f"加密套件: {info.cipher}")

5. 环境变量与配置管理

5.1 配置分层设计

# config/base.py
"""
基础配置类
"""

from pydantic_settings import BaseSettings
from functools import lru_cache
from typing import List
import os


class BaseConfig(BaseSettings):
    """基础配置"""
    
    # 应用信息
    APP_NAME: str = "MyApp"
    APP_VERSION: str = "1.0.0"
    DEBUG: bool = False
    
    # 服务器配置
    HOST: str = "0.0.0.0"
    PORT: int = 8000
    WORKERS: int = 4
    
    # 安全配置
    SECRET_KEY: str = "change-me-in-production"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    
    # 数据库配置
    DATABASE_URL: str = "sqlite:///./app.db"
    DATABASE_POOL_SIZE: int = 20
    DATABASE_MAX_OVERFLOW: int = 10
    
    # Redis 配置
    REDIS_URL: str = "redis://localhost:6379/0"
    REDIS_POOL_SIZE: int = 50
    
    # 日志配置
    LOG_LEVEL: str = "INFO"
    LOG_FORMAT: str = "json"  # json 或 text
    
    # CORS 配置
    CORS_ORIGINS: List[str] = ["http://localhost:3000"]
    CORS_ALLOW_CREDENTIALS: bool = True
    
    # 速率限制
    RATE_LIMIT_PER_MINUTE: int = 60
    
    class Config:
        env_file = ".env"
        case_sensitive = True


# config/development.py
class DevelopmentConfig(BaseConfig):
    """开发环境配置"""
    DEBUG: bool = True
    LOG_LEVEL: str = "DEBUG"
    DATABASE_URL: str = "sqlite:///./dev.db"
    CORS_ORIGINS: List[str] = ["*"]


# config/production.py
class ProductionConfig(BaseConfig):
    """生产环境配置"""
    DEBUG: bool = False
    LOG_LEVEL: str = "INFO"
    LOG_FORMAT: str = "json"
    
    # 生产环境必须使用强密钥
    SECRET_KEY: str  # 必须设置
    
    # 生产数据库
    DATABASE_URL: str  # 必须设置
    
    # 严格的安全配置
    CORS_ORIGINS: List[str] = []  # 必须明确设置
    CORS_ALLOW_CREDENTIALS: bool = True
    
    # 更严格的速率限制
    RATE_LIMIT_PER_MINUTE: int = 30


# config/testing.py
class TestingConfig(BaseConfig):
    """测试环境配置"""
    DEBUG: bool = True
    TESTING: bool = True
    DATABASE_URL: str = "sqlite:///./test.db"
    
    # 测试时禁用某些功能
    ENABLE_RATE_LIMIT: bool = False
    ENABLE_CACHE: bool = False


# config/__init__.py
import os

# 配置映射
config_map = {
    "development": DevelopmentConfig,
    "production": ProductionConfig,
    "testing": TestingConfig,
}


def get_config():
    """获取当前环境配置"""
    env = os.getenv("APP_ENV", "development")
    config_class = config_map.get(env, DevelopmentConfig)
    return config_class()


@lru_cache()
def get_settings():
    """获取缓存的配置实例"""
    return get_config()

5.2 环境变量文件管理

# .env.example - 环境变量模板（提交到版本控制）
# 复制此文件为 .env 并填入实际值

# 应用环境: development | production | testing
APP_ENV=development

# 安全密钥（生产环境必须修改！）
SECRET_KEY=your-super-secret-key-here

# 数据库
DATABASE_URL=postgresql://user:password@localhost/dbname

# Redis
REDIS_URL=redis://localhost:6379/0

# 外部服务 API 密钥
SENDGRID_API_KEY=
AWS_ACCESS_KEY_ID=
AWS_SECRET_ACCESS_KEY=

# 日志
LOG_LEVEL=INFO
SENTRY_DSN=

# .env - 实际环境变量文件（不提交到版本控制）
APP_ENV=production
SECRET_KEY=your-256-bit-secret-key-here
DATABASE_URL=postgresql://appuser:secure_password@db.example.com:5432/myapp
REDIS_URL=redis://redis.example.com:6379/0
LOG_LEVEL=WARNING

5.3 Docker 环境配置

# Dockerfile
# 生产环境 Dockerfile

FROM python:3.12-slim as builder

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

# 创建虚拟环境
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

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

# 生产镜像
FROM python:3.12-slim

# 安装运行时依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    libpq5 \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 从构建阶段复制虚拟环境
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# 创建非 root 用户
RUN useradd --create-home --shell /bin/bash app
WORKDIR /home/app
USER app

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

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["gunicorn", "-c", "gunicorn.conf.py", "main:app"]

# docker-compose.yml
# 生产环境 Docker Compose 配置

version: '3.8'

services:
  app:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: myapp
    restart: unless-stopped
    environment:
      - APP_ENV=production
      - DATABASE_URL=postgresql://postgres:password@db:5432/myapp
      - REDIS_URL=redis://redis:6379/0
      - SECRET_KEY=${SECRET_KEY}
    ports:
      - "8000:8000"
    depends_on:
      - db
      - redis
    volumes:
      - ./static:/home/app/static
      - ./media:/home/app/media
      - ./logs:/home/app/logs
    networks:
      - app-network
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 40s

  db:
    image: postgres:16-alpine
    container_name: myapp-db
    restart: unless-stopped
    environment:
      - POSTGRES_DB=myapp
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    networks:
      - app-network
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    container_name: myapp-redis
    restart: unless-stopped
    volumes:
      - redis_data:/data
    networks:
      - app-network
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 3s
      retries: 5

  nginx:
    image: nginx:alpine
    container_name: myapp-nginx
    restart: unless-stopped
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
      - ./nginx/sites-available:/etc/nginx/sites-available:ro
      - ./static:/var/www/static:ro
      - ./media:/var/www/media:ro
      - ./certbot/conf:/etc/letsencrypt:ro
      - ./certbot/www:/var/www/certbot:ro
    depends_on:
      - app
    networks:
      - app-network

volumes:
  postgres_data:
  redis_data:

networks:
  app-network:
    driver: bridge

5.4 密钥管理服务集成

# secrets_manager.py
"""
密钥管理服务集成
支持 AWS Secrets Manager、Azure Key Vault、HashiCorp Vault
"""

import os
import json
from abc import ABC, abstractmethod
from typing import Dict, Any


class SecretsProvider(ABC):
    """密钥提供者抽象基类"""
    
    @abstractmethod
    def get_secret(self, secret_name: str) -> Dict[str, Any]:
        """获取密钥"""
        pass


class AWSSecretsManager(SecretsProvider):
    """AWS Secrets Manager 实现"""
    
    def __init__(self, region_name: str = "us-east-1"):
        import boto3
        self.client = boto3.client(
            service_name='secretsmanager',
            region_name=region_name
        )
    
    def get_secret(self, secret_name: str) -> Dict[str, Any]:
        """从 AWS Secrets Manager 获取密钥"""
        try:
            response = self.client.get_secret_value(SecretId=secret_name)
            if 'SecretString' in response:
                return json.loads(response['SecretString'])
            else:
                import base64
                return json.loads(base64.b64decode(response['SecretBinary']))
        except Exception as e:
            raise RuntimeError(f"无法获取密钥 {secret_name}: {e}")


class AzureKeyVault(SecretsProvider):
    """Azure Key Vault 实现"""
    
    def __init__(self, vault_url: str):
        from azure.identity import DefaultAzureCredential
        from azure.keyvault.secrets import SecretClient
        
        credential = DefaultAzureCredential()
        self.client = SecretClient(vault_url=vault_url, credential=credential)
    
    def get_secret(self, secret_name: str) -> Dict[str, Any]:
        """从 Azure Key Vault 获取密钥"""
        try:
            secret = self.client.get_secret(secret_name)
            return json.loads(secret.value)
        except Exception as e:
            raise RuntimeError(f"无法获取密钥 {secret_name}: {e}")


class HashiCorpVault(SecretsProvider):
    """HashiCorp Vault 实现"""
    
    def __init__(self, url: str, token: str = None):
        import hvac
        
        self.client = hvac.Client(url=url)
        if token:
            self.client.token = token
        elif not self.client.is_authenticated():
            raise RuntimeError("Vault 未认证")
    
    def get_secret(self, secret_path: str) -> Dict[str, Any]:
        """从 HashiCorp Vault 获取密钥"""
        try:
            response = self.client.secrets.kv.v2.read_secret_version(
                path=secret_path
            )
            return response['data']['data']
        except Exception as e:
            raise RuntimeError(f"无法获取密钥 {secret_path}: {e}")


def load_secrets_from_provider() -> Dict[str, Any]:
    """从配置的密钥提供者加载密钥"""
    provider_type = os.getenv('SECRETS_PROVIDER', 'env')
    
    if provider_type == 'aws':
        provider = AWSSecretsManager(
            region_name=os.getenv('AWS_REGION', 'us-east-1')
        )
        secret_name = os.getenv('AWS_SECRET_NAME')
        return provider.get_secret(secret_name)
    
    elif provider_type == 'azure':
        provider = AzureKeyVault(
            vault_url=os.getenv('AZURE_VAULT_URL')
        )
        secret_name = os.getenv('AZURE_SECRET_NAME')
        return provider.get_secret(secret_name)
    
    elif provider_type == 'vault':
        provider = HashiCorpVault(
            url=os.getenv('VAULT_URL'),
            token=os.getenv('VAULT_TOKEN')
        )
        secret_path = os.getenv('VAULT_SECRET_PATH')
        return provider.get_secret(secret_path)
    
    else:
        # 默认从环境变量读取
        return dict(os.environ)


# 在配置中使用
# config = load_secrets_from_provider()
# DATABASE_URL = config.get('DATABASE_URL')

6. 完整部署流程

6.1 部署脚本

#!/bin/bash
# deploy.sh
# 完整的生产环境部署脚本

set -e  # 遇到错误立即退出

# 配置
APP_NAME="myapp"
APP_DIR="/opt/${APP_NAME}"
USER="appuser"
DOMAIN="example.com"

# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color

log() {
    echo -e "${GREEN}[INFO]${NC} $1"
}

warn() {
    echo -e "${YELLOW}[WARN]${NC} $1"
}

error() {
    echo -e "${RED}[ERROR]${NC} $1"
    exit 1
}

# 检查 root 权限
if [ "$EUID" -ne 0 ]; then
    error "请使用 sudo 运行此脚本"
fi

log "开始部署 ${APP_NAME}..."

# 1. 系统更新
log "更新系统包..."
apt-get update && apt-get upgrade -y

# 2. 安装依赖
log "安装系统依赖..."
apt-get install -y \
    python3.12 \
    python3.12-venv \
    python3-pip \
    nginx \
    postgresql \
    postgresql-contrib \
    redis-server \
    git \
    curl \
    certbot \
    python3-certbot-nginx \
    supervisor

# 3. 创建应用用户
if ! id "$USER" &>/dev/null; then
    log "创建应用用户..."
    useradd -m -s /bin/bash "$USER"
fi

# 4. 创建目录结构
log "创建目录结构..."
mkdir -p "${APP_DIR}"/{static,media,logs,backup}
chown -R "${USER}:${USER}" "${APP_DIR}"

# 5. 克隆/更新代码
log "部署应用代码..."
if [ -d "${APP_DIR}/.git" ]; then
    cd "${APP_DIR}"
    sudo -u "$USER" git pull origin main
else
    sudo -u "$USER" git clone https://github.com/yourusername/myapp.git "${APP_DIR}"
fi

# 6. 创建虚拟环境
log "创建 Python 虚拟环境..."
cd "${APP_DIR}"
python3.12 -m venv venv
source venv/bin/activate
pip install --upgrade pip
pip install -r requirements.txt

# 7. 配置环境变量
log "配置环境变量..."
if [ ! -f "${APP_DIR}/.env" ]; then
    warn "请手动创建 ${APP_DIR}/.env 文件"
    cp "${APP_DIR}/.env.example" "${APP_DIR}/.env"
fi

# 8. 数据库迁移
log "执行数据库迁移..."
cd "${APP_DIR}"
source venv/bin/activate
alembic upgrade head

# 9. 收集静态文件
log "收集静态文件..."
python -c "import shutil; shutil.copytree('static', '${APP_DIR}/static', dirs_exist_ok=True)"

# 10. 配置 Supervisor
log "配置 Supervisor..."
cat > /etc/supervisor/conf.d/${APP_NAME}.conf << EOF
[program:${APP_NAME}]
directory=${APP_DIR}
command=${APP_DIR}/venv/bin/gunicorn -c gunicorn.conf.py main:app
user=${USER}
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=${APP_DIR}/logs/supervisor.log
environment=APP_ENV="production"
EOF

supervisorctl reread
supervisorctl update
supervisorctl restart "${APP_NAME}"

# 11. 配置 Nginx
log "配置 Nginx..."
cp "${APP_DIR}/nginx/myapp" /etc/nginx/sites-available/
ln -sf /etc/nginx/sites-available/myapp /etc/nginx/sites-enabled/
rm -f /etc/nginx/sites-enabled/default

nginx -t && systemctl reload nginx

# 12. 配置 SSL（如果需要）
if [ ! -d "/etc/letsencrypt/live/${DOMAIN}" ]; then
    log "配置 SSL 证书..."
    certbot --nginx -d "${DOMAIN}" -d "www.${DOMAIN}" --non-interactive --agree-tos --email admin@${DOMAIN}
fi

# 13. 设置防火墙
log "配置防火墙..."
ufw allow 'Nginx Full'
ufw allow OpenSSH
ufw --force enable

# 14. 设置自动备份
log "配置自动备份..."
(crontab -u "$USER" -l 2>/dev/null; echo "0 2 * * * ${APP_DIR}/scripts/backup.sh") | crontab -u "$USER" -

log "部署完成！"
log "应用地址: https://${DOMAIN}"
log "健康检查: https://${DOMAIN}/health"

6.2 健康检查端点

# health_check.py
"""
健康检查端点实现
"""

from fastapi import FastAPI, Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
import asyncio
import redis.asyncio as redis
from datetime import datetime
from typing import Dict, Any

app = FastAPI()

# Redis 客户端
redis_client = redis.Redis(host='localhost', port=6379, db=0)


async def check_database(db: AsyncSession) -> Dict[str, Any]:
    """检查数据库连接"""
    try:
        start = datetime.now()
        await db.execute(text("SELECT 1"))
        latency = (datetime.now() - start).total_seconds() * 1000
        return {
            "status": "healthy",
            "latency_ms": round(latency, 2)
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "error": str(e)
        }


async def check_redis() -> Dict[str, Any]:
    """检查 Redis 连接"""
    try:
        start = datetime.now()
        await redis_client.ping()
        latency = (datetime.now() - start).total_seconds() * 1000
        return {
            "status": "healthy",
            "latency_ms": round(latency, 2)
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "error": str(e)
        }


async def check_disk_space() -> Dict[str, Any]:
    """检查磁盘空间"""
    import shutil
    
    try:
        total, used, free = shutil.disk_usage("/")
        usage_percent = (used / total) * 100
        
        status = "healthy"
        if usage_percent > 90:
            status = "critical"
        elif usage_percent > 80:
            status = "warning"
        
        return {
            "status": status,
            "total_gb": round(total / (1024**3), 2),
            "used_gb": round(used / (1024**3), 2),
            "free_gb": round(free / (1024**3), 2),
            "usage_percent": round(usage_percent, 2)
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "error": str(e)
        }


@app.get("/health")
async def health_check():
    """简单健康检查"""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat()
    }


@app.get("/health/detailed")
async def detailed_health_check():
    """详细健康检查"""
    checks = await asyncio.gather(
        check_redis(),
        check_disk_space(),
        return_exceptions=True
    )
    
    results = {
        "redis": checks[0] if not isinstance(checks[0], Exception) else {"status": "unhealthy", "error": str(checks[0])},
        "disk": checks[1] if not isinstance(checks[1], Exception) else {"status": "unhealthy", "error": str(checks[1])},
    }
    
    # 总体状态
    overall_status = "healthy"
    for check in results.values():
        if check.get("status") == "critical":
            overall_status = "critical"
            break
        elif check.get("status") == "unhealthy":
            overall_status = "unhealthy"
        elif check.get("status") == "warning" and overall_status == "healthy":
            overall_status = "warning"
    
    return {
        "status": overall_status,
        "timestamp": datetime.now().isoformat(),
        "checks": results
    }


@app.get("/ready")
async def readiness_check():
    """就绪检查（用于 Kubernetes）"""
    # 检查应用是否准备好接收流量
    return {"ready": True}


@app.get("/live")
async def liveness_check():
    """存活检查（用于 Kubernetes）"""
    # 检查应用是否还在运行
    return {"alive": True}

避坑小贴士

1. 不要在生产环境使用开发服务器

# 错误：使用 Flask/FastAPI 开发服务器
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)  # 只适合开发！

# 正确：使用 Gunicorn + Uvicorn
gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app

2. 注意环境变量的安全性

# 错误：将敏感信息硬编码
DATABASE_PASSWORD = "mysecretpassword"  # 永远不要这样做！

# 正确：从环境变量读取
import os
DATABASE_PASSWORD = os.getenv("DATABASE_PASSWORD")
if not DATABASE_PASSWORD:
    raise ValueError("DATABASE_PASSWORD 环境变量未设置")

3. 静态文件配置问题

# 错误：使用 root 而不是 alias
location /static/ {
    root /opt/myapp/static/;  # 这会导致路径变成 /opt/myapp/static/static/
}

# 正确：使用 alias
location /static/ {
    alias /opt/myapp/static/;  # 正确映射到 /opt/myapp/static/
}

4. 数据库连接池耗尽

# 问题：连接池配置不当导致连接耗尽
engine = create_async_engine(
    DATABASE_URL,
    pool_size=100,  # 太大
    max_overflow=100  # 太大
)

# 正确：根据实际负载配置
engine = create_async_engine(
    DATABASE_URL,
    pool_size=10,  # 基础连接数
    max_overflow=20,  # 额外连接数
    pool_pre_ping=True,  # 连接前检查
    pool_recycle=3600  # 定期回收连接
)

5. SSL 证书续期失败

# 问题：Certbot 续期失败导致证书过期
# 解决方案：

# 1. 测试续期
sudo certbot renew --dry-run

# 2. 确保 Nginx 配置正确
sudo nginx -t

# 3. 检查防火墙是否允许 80 端口
sudo ufw status

# 4. 手动续期并重启 Nginx
sudo certbot renew --quiet --deploy-hook "systemctl reload nginx"

课后练习

练习 1：基础部署

在本地虚拟机或云服务器上部署一个 FastAPI 应用
配置 Gunicorn + Uvicorn 运行
配置 Nginx 反向代理
申请并配置 Let’s Encrypt SSL 证书

练习 2：性能调优

使用 wrk 或 ab 对应用进行压力测试
调整 Gunicorn 工作进程数，观察性能变化
开启 Nginx Gzip 压缩，对比响应大小
配置静态文件缓存，测试缓存效果

练习 3：高可用部署

在同一台服务器上启动多个应用实例（不同端口）
配置 Nginx 负载均衡
实现健康检查和自动故障转移
编写自动化部署脚本

练习 4：安全配置

配置所有安全响应头
实现 API 速率限制
配置防火墙规则
使用密钥管理服务存储敏感配置

下一篇预告

第24讲：云原生与 CI/CD

在下一讲中，我们将学习：

Docker 容器化最佳实践
Kubernetes 基础与应用部署
GitHub Actions CI/CD 流水线
监控、日志收集与告警

云原生技术正在改变应用部署的方式，敬请期待！

参考资源

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

深入理解 AI Agent · AGENT #02：Agent 的四大核心机制，拆开来看每个都在干什么

AI Agent技术社区

MCP协议开发实战：从零搭建AI Agent工具链

本文介绍了MCP协议在AI工具链中的应用与实践。MCP协议旨在解决异构工具集成难题，为AI模型提供标准化、安全的工具调用接口。文章详细解析了MCP的核心架构，包括Server、Client、Transport等组件，以及通信流程与安全模型。通过实战示例，指导读者开发计算器和文件系统工具Server，并集成到Claude等AI客户端中。进阶部分展示了如何构建复杂工具链，包括API集成、数据库查询和多