云部署实战

# config/database.py — FastAPI 数据库配置
from dataclasses import dataclass
import os


@dataclass
class DatabaseConfig:
    """RDS 数据库连接配置"""
    host: str = os.getenv("RDS_HOST", "localhost")
    port: int = int(os.getenv("RDS_PORT", "5432"))
    user: str = os.getenv("RDS_USER", "postgres")
    password: str = os.getenv("RDS_PASSWORD", "")
    database: str = os.getenv("RDS_DATABASE", "app_db")

    @property
    def dsn(self) -> str:
        """SQLAlchemy 连接字符串"""
        return (
            f"postgresql+asyncpg://{self.user}:{self.password}"
            f"@{self.host}:{self.port}/{self.database}"
        )

    @property
    def dsn_sync(self) -> str:
        """同步连接字符串（用于迁移）"""
        return (
            f"postgresql+psycopg2://{self.user}:{self.password}"
            f"@{self.host}:{self.port}/{self.database}"
        )


# 生产环境推荐配置
# - 实例类型: db.t3.medium（开发）/ db.r6g.large（生产）
# - 多可用区部署: 启用 Multi-AZ
# - 自动备份: 保留 7-30 天
# - 加密: 启用存储加密（KMS）
# - 安全组: 仅允许 EC2 实例安全组访问 5432 端口

# storage/s3_client.py — S3 文件上传/下载
import boto3
from botocore.config import Config as BotoConfig
from dataclasses import dataclass
from pathlib import Path


@dataclass
class S3Config:
    bucket: str = os.getenv("S3_BUCKET", "my-app-uploads")
    region: str = os.getenv("AWS_REGION", "us-east-1")
    endpoint_url: str | None = os.getenv("S3_ENDPOINT_URL")  # 本地开发用 MinIO


class S3Client:
    """S3 文件操作封装"""

    def __init__(self, config: S3Config | None = None) -> None:
        self.config = config or S3Config()
        self.client = boto3.client(
            "s3",
            region_name=self.config.region,
            endpoint_url=self.config.endpoint_url,
            config=BotoConfig(
                retries={"max_attempts": 3, "mode": "standard"},
                connect_timeout=5,
                read_timeout=30,
            ),
        )

    def upload_file(self, local_path: str, s3_key: str) -> str:
        """上传文件到 S3，返回公开 URL"""
        extra_args = {
            "ContentType": self._guess_content_type(local_path),
        }

        self.client.upload_file(
            local_path,
            self.config.bucket,
            s3_key,
            ExtraArgs=extra_args,
        )

        return f"https://{self.config.bucket}.s3.{self.config.region}.amazonaws.com/{s3_key}"

    def upload_bytes(self, data: bytes, s3_key: str, content_type: str = "application/octet-stream") -> str:
        """上传二进制数据到 S3"""
        self.client.put_object(
            Bucket=self.config.bucket,
            Key=s3_key,
            Body=data,
            ContentType=content_type,
        )
        return f"https://{self.config.bucket}.s3.{self.config.region}.amazonaws.com/{s3_key}"

    def download_file(self, s3_key: str, local_path: str) -> None:
        """从 S3 下载文件"""
        Path(local_path).parent.mkdir(parents=True, exist_ok=True)
        self.client.download_file(self.config.bucket, s3_key, local_path)

    def generate_presigned_url(self, s3_key: str, expires_in: int = 3600) -> str:
        """生成预签名 URL（临时访问）"""
        return self.client.generate_presigned_url(
            "get_object",
            Params={"Bucket": self.config.bucket, "Key": s3_key},
            ExpiresIn=expires_in,
        )

    def delete_file(self, s3_key: str) -> None:
        """删除 S3 文件"""
        self.client.delete_object(Bucket=self.config.bucket, Key=s3_key)

    @staticmethod
    def _guess_content_type(path: str) -> str:
        """根据扩展名推断 Content-Type"""
        import mimetypes
        content_type, _ = mimetypes.guess_type(path)
        return content_type or "application/octet-stream"


import os  # noqa: E402 — S3Config 需要 os

# monitoring/cloudwatch.py — CloudWatch 指标上报与日志
import boto3
import logging
from datetime import datetime, timezone
from typing import Any


logger = logging.getLogger(__name__)


class CloudWatchMetrics:
    """CloudWatch 自定义指标上报"""

    def __init__(self, namespace: str = "MyApp") -> None:
        self.client = boto3.client("cloudwatch")
        self.namespace = namespace

    def put_metric(
        self,
        metric_name: str,
        value: float,
        unit: str = "Count",
        dimensions: dict[str, str] | None = None,
    ) -> None:
        """上报单个指标"""
        metric_data = {
            "MetricName": metric_name,
            "Value": value,
            "Unit": unit,
            "Timestamp": datetime.now(timezone.utc),
        }

        if dimensions:
            metric_data["Dimensions"] = [
                {"Name": k, "Value": v} for k, v in dimensions.items()
            ]

        try:
            self.client.put_metric_data(
                Namespace=self.namespace,
                MetricData=[metric_data],
            )
        except Exception as e:
            logger.error(f"CloudWatch 上报失败: {e}")

    def put_metrics_batch(self, metrics: list[dict[str, Any]]) -> None:
        """批量上报指标（最多 20 个）"""
        try:
            self.client.put_metric_data(
                Namespace=self.namespace,
                MetricData=metrics[:20],
            )
        except Exception as e:
            logger.error(f"CloudWatch 批量上报失败: {e}")


class CloudWatchLogger:
    """CloudWatch Logs 日志推送"""

    def __init__(self, log_group: str, log_stream: str) -> None:
        self.client = boto3.client("logs")
        self.log_group = log_group
        self.log_stream = log_stream
        self._sequence_token: str | None = None
        self._ensure_log_stream()

    def _ensure_log_stream(self) -> None:
        """确保日志流存在"""
        try:
            self.client.create_log_stream(
                logGroupName=self.log_group,
                logStreamName=self.log_stream,
            )
        except self.client.exceptions.ResourceAlreadyExistsException:
            pass

    def log(self, message: str) -> None:
        """发送日志到 CloudWatch"""
        try:
            kwargs = {
                "logGroupName": self.log_group,
                "logStreamName": self.log_stream,
                "logEvents": [
                    {
                        "timestamp": int(datetime.now(timezone.utc).timestamp() * 1000),
                        "message": message,
                    }
                ],
            }

            if self._sequence_token:
                kwargs["sequenceToken"] = self._sequence_token

            response = self.client.put_log_events(**kwargs)
            self._sequence_token = response.get("nextSequenceToken")
        except Exception as e:
            logger.error(f"CloudWatch 日志推送失败: {e}")

# lambda/handler.py — AWS Lambda 处理函数
import json
import logging
from dataclasses import dataclass
from typing import Any

logger = logging.getLogger()
logger.setLevel(logging.INFO)


@dataclass
class APIResponse:
    """API Gateway 响应封装"""
    status_code: int = 200
    body: dict | list | str = ""
    headers: dict[str, str] | None = None

    def to_dict(self) -> dict[str, Any]:
        default_headers = {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Methods": "GET, POST, OPTIONS",
            "Access-Control-Allow-Headers": "Content-Type, Authorization",
        }
        if self.headers:
            default_headers.update(self.headers)

        body = self.body if isinstance(self.body, str) else json.dumps(self.body, ensure_ascii=False)

        return {
            "statusCode": self.status_code,
            "body": body,
            "headers": default_headers,
        }


def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    """Lambda 入口函数

    event 结构取决于触发器类型：
    - API Gateway: event['httpMethod'], event['pathParameters'], event['body']
    - S3: event['Records'][0]['s3']
    - SQS: event['Records'][0]['body']
    - EventBridge: event['detail']
    """
    logger.info(f"收到事件: {json.dumps(event, default=str)}")

    try:
        http_method = event.get("httpMethod", "GET")
        path = event.get("path", "/")
        body = event.get("body", "{}")

        # 解析请求体
        if isinstance(body, str):
            try:
                request_data = json.loads(body)
            except json.JSONDecodeError:
                request_data = {}
        else:
            request_data = body

        # 路由分发
        if path == "/api/hello" and http_method == "GET":
            return _handle_hello(request_data)
        elif path == "/api/data" and http_method == "POST":
            return _handle_create_data(request_data)
        elif path == "/api/data" and http_method == "GET":
            return _handle_list_data(event)
        else:
            return APIResponse(status_code=404, body={"error": "Not Found"}).to_dict()

    except Exception as e:
        logger.exception("处理请求异常")
        return APIResponse(
            status_code=500,
            body={"error": "Internal Server Error", "detail": str(e)},
        ).to_dict()


def _handle_hello(data: dict) -> dict[str, Any]:
    """GET /api/hello"""
    name = data.get("name", "World")
    return APIResponse(body={"message": f"Hello, {name}!"}).to_dict()


def _handle_create_data(data: dict) -> dict[str, Any]:
    """POST /api/data"""
    # 实际项目中这里会写入 DynamoDB 或 RDS
    logger.info(f"创建数据: {data}")
    return APIResponse(
        status_code=201,
        body={"message": "Created", "data": data},
    ).to_dict()


def _handle_list_data(event: dict) -> dict[str, Any]:
    """GET /api/data"""
    # 实际项目中这里会从数据库查询
    query_params = event.get("queryStringParameters") or {}
    page = int(query_params.get("page", "1"))
    size = int(query_params.get("size", "10"))

    return APIResponse(body={
        "items": [],
        "page": page,
        "size": size,
        "total": 0,
    }).to_dict()

# fc/handler.py — 阿里云函数计算入口
import json
import logging
import os

logger = logging.getLogger()

# 阿里云函数计算使用 WSGI 兼容接口（自定义运行时）
# 或事件触发接口（事件函数）


def handler(event: dict, context: dict) -> dict:
    """阿里云函数计算 — 事件触发入口

    event: 触发器传入的事件数据
    context: 运行时上下文，包含 request_id, credentials 等
    """
    logger.info(f"FC 收到事件: {json.dumps(event, default=str)}")

    # 获取运行时信息
    request_id = context.get("requestId", "unknown")
    region = os.getenv("FC_REGION", "cn-hangzhou")

    # 解析触发器类型
    trigger_type = _detect_trigger(event)

    if trigger_type == "http":
        return _handle_http(event, request_id)
    elif trigger_type == "oss":
        return _handle_oss_event(event, request_id)
    elif trigger_type == "timer":
        return _handle_timer(event, request_id)
    else:
        return {"statusCode": 200, "body": json.dumps({"message": "OK"})}


def _detect_trigger(event: dict) -> str:
    """检测触发器类型"""
    if "httpMethod" in event or "method" in event:
        return "http"
    elif "events" in event:
        return "oss"
    elif "triggerName" in event:
        return "timer"
    return "unknown"


def _handle_http(event: dict, request_id: str) -> dict:
    """处理 HTTP 触发"""
    method = event.get("method", event.get("httpMethod", "GET"))
    path = event.get("path", "/")
    query = event.get("queries", {})
    body = event.get("body", "{}")

    if isinstance(body, str):
        try:
            data = json.loads(body)
        except json.JSONDecodeError:
            data = {}
    else:
        data = body

    logger.info(f"HTTP {method} {path} [request_id={request_id}]")

    return {
        "isBase64Encoded": False,
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({
            "message": "Hello from Alibaba Cloud FC!",
            "method": method,
            "path": path,
            "request_id": request_id,
        }, ensure_ascii=False),
    }


def _handle_oss_event(event: dict, request_id: str) -> dict:
    """处理 OSS 事件触发"""
    for evt in event.get("events", []):
        bucket = evt.get("oss", {}).get("bucket", {}).get("name")
        key = evt.get("oss", {}).get("object", {}).get("key")
        logger.info(f"OSS 事件: bucket={bucket}, key={key}, request_id={request_id}")

    return {"statusCode": 200, "body": "OK"}


def _handle_timer(event: dict, request_id: str) -> dict:
    """处理定时触发"""
    trigger_name = event.get("triggerName", "unknown")
    logger.info(f"定时触发: {trigger_name}, request_id={request_id}")
    return {"statusCode": 200, "body": "OK"}

# serverless.yml — Serverless Framework 部署配置
service: python-api

frameworkVersion: "3"

provider:
  name: aws
  runtime: python3.12
  region: us-east-1
  stage: ${opt:stage, "dev"}
  timeout: 30            # 函数超时（秒）
  memorySize: 256        # 内存（MB）
  logRetentionInDays: 14 # CloudWatch 日志保留天数

  # 环境变量
  environment:
    STAGE: ${self:provider.stage}
    RDS_HOST: ${env:RDS_HOST}
    RDS_PORT: "5432"
    RDS_DATABASE: app_db

  # IAM 权限
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - s3:GetObject
            - s3:PutObject
          Resource: "arn:aws:s3:::my-app-uploads/*"
        - Effect: Allow
          Action:
            - dynamodb:GetItem
            - dynamodb:PutItem
            - dynamodb:Query
          Resource: "arn:aws:dynamodb:${self:provider.region}:*:table/app-*"

functions:
  api:
    handler: lambda/handler.lambda_handler
    events:
      - http:
          path: /api/{proxy+}
          method: ANY
          cors: true

  # S3 事件触发函数
  processUpload:
    handler: lambda/process_upload.handler
    events:
      - s3:
          bucket: my-app-uploads
          event: s3:ObjectCreated:*
          existing: true

  # 定时任务
  scheduledCleanup:
    handler: lambda/cleanup.handler
    events:
      - schedule:
          rate: cron(0 2 * * ? *)
          description: "每天凌晨 2 点清理过期数据"

# 资源（CloudFormation）
resources:
  Resources:
    # DynamoDB 表
    AppTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: app-data-${self:provider.stage}
        BillingMode: PAY_PER_REQUEST
        AttributeDefinitions:
          - AttributeName: pk
            AttributeType: S
          - AttributeName: sk
            AttributeType: S
        KeySchema:
          - AttributeName: pk
            KeyType: HASH
          - AttributeName: sk
            KeyType: RANGE

plugins:
  - serverless-python-requirements

custom:
  pythonRequirements:
    dockerizePip: true    # 非 Linux 环境用 Docker 打包依赖
    slim: true            # 移除 .pyc 和 dist-info
    layer: true           # 依赖作为 Lambda Layer

#!/bin/bash
# scripts/setup-https.sh — EC2 上配置 Let's Encrypt HTTPS

set -euo pipefail

DOMAIN="${1:?用法: $0 <domain>}"
EMAIL="${2:?用法: $0 <domain> <email>}"

echo "=== 为 ${DOMAIN} 配置 HTTPS ==="

# 1. 安装 Certbot
echo "[1/4] 安装 Certbot..."
sudo apt-get update -qq
sudo apt-get install -y certbot python3-certbot-nginx

# 2. 获取证书（Nginx 插件）
echo "[2/4] 获取 SSL 证书..."
sudo certbot --nginx \
  --domain "${DOMAIN}" \
  --domain "www.${DOMAIN}" \
  --non-interactive \
  --agree-tos \
  --email "${EMAIL}" \
  --redirect

# 3. 配置自动续期
echo "[3/4] 配置自动续期..."
sudo crontab -l 2>/dev/null | grep -v certbot || true
echo "0 3 * * * certbot renew --quiet --post-hook 'systemctl reload nginx'" | sudo crontab -

# 4. 验证
echo "[4/4] 验证证书..."
sudo certbot certificates

echo ""
echo "=== HTTPS 配置完成 ==="
echo "访问 https://${DOMAIN} 验证"
echo "自动续期: 每天凌晨 3 点检查"

#!/bin/bash
# scripts/setup-cloudflare.sh — CloudFlare DNS 代理配置

# CloudFlare 提供:
# 1. 免费 DNS 解析（Anycast 全球节点）
# 2. 免费 SSL（Universal SSL）
# 3. 免费 CDN 缓存
# 4. DDoS 防护
# 5. WAF（付费）

# 配置步骤:

# 步骤1: 在 CloudFlare 添加域名
# 通过 CloudFlare Dashboard 或 API:

CF_API_TOKEN="${CF_API_TOKEN:?请设置 CF_API_TOKEN}"
CF_ZONE_ID="${CF_ZONE_ID:?请设置 CF_ZONE_ID}"
DOMAIN="example.com"
EC2_IP="1.2.3.4"

# 步骤2: 添加 DNS A 记录（橙色云朵 = 代理模式）
curl -s -X POST "https://api.cloudflare.com/client/v4/zones/${CF_ZONE_ID}/dns_records" \
  -H "Authorization: Bearer ${CF_API_TOKEN}" \
  -H "Content-Type: application/json" \
  --data "{
    \"type\": \"A\",
    \"name\": \"${DOMAIN}\",
    \"content\": \"${EC2_IP}\",
    \"ttl\": 1,
    \"proxied\": true
  }" | jq .

# 步骤3: 添加 www CNAME
curl -s -X POST "https://api.cloudflare.com/client/v4/zones/${CF_ZONE_ID}/dns_records" \
  -H "Authorization: Bearer ${CF_API_TOKEN}" \
  -H "Content-Type: application/json" \
  --data "{
    \"type\": \"CNAME\",
    \"name\": \"www.${DOMAIN}\",
    \"content\": \"${DOMAIN}\",
    \"ttl\": 1,
    \"proxied\": true
  }" | jq .

# 步骤4: 设置 SSL 模式为 Full (Strict)
# Dashboard: SSL/TLS → Overview → Full (Strict)
# 或通过 API:
curl -s -X PATCH "https://api.cloudflare.com/client/v4/zones/${CF_ZONE_ID}/settings/ssl" \
  -H "Authorization: Bearer ${CF_API_TOKEN}" \
  -H "Content-Type: application/json" \
  --data '{"value": "strict"}' | jq .

echo "CloudFlare 配置完成"
echo "DNS 生效通常需要几分钟到 24 小时"

# /etc/nginx/sites-available/fastapi-app — Nginx HTTPS 反向代理配置

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

# HTTPS 服务
server {
    listen 443 ssl http2;
    server_name example.com www.example.com;

    # SSL 证书（Let's Encrypt）
    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_ciphers ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384;
    ssl_prefer_server_ciphers off;

    # HSTS（强制 HTTPS，有效期 1 年）
    add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" 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;

    # 日志
    access_log /var/log/nginx/fastapi_access.log;
    error_log /var/log/nginx/fastapi_error.log;

    # 反向代理到 FastAPI（Docker 容器）
    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;

        # WebSocket 支持
        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 300s;  # 长轮询/大文件上传需要更长超时
    }

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

    # 健康检查端点（不记日志）
    location /health {
        proxy_pass http://127.0.0.1:8000/health;
        access_log off;
    }
}

my-fastapi-app/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI 应用
│   ├── config.py         # 配置
│   ├── models.py         # 数据模型
│   ├── routers/
│   │   ├── __init__.py
│   │   └── api.py
│   └── database.py       # 数据库连接
├── Dockerfile
├── docker-compose.yml
├── nginx/
│   └── nginx.conf
├── scripts/
│   ├── deploy.sh         # 部署脚本
│   └── setup-https.sh    # HTTPS 配置
├── .env.example
├── requirements.txt
└── pyproject.toml

# app/main.py — FastAPI 应用入口
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import logging

from app.database import init_db, close_db
from app.routers import api

logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    logger.info("应用启动中...")
    await init_db()
    yield
    logger.info("应用关闭中...")
    await close_db()


app = FastAPI(
    title="My FastAPI App",
    version="1.0.0",
    lifespan=lifespan,
)

# CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],  # 生产环境限制具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 路由
app.include_router(api.router, prefix="/api/v1")


@app.get("/health")
async def health_check():
    """健康检查端点（供 ALB/CloudWatch 探测）"""
    return {"status": "healthy", "version": "1.0.0"}

# Dockerfile — 多阶段构建，优化镜像大小

# ---- 构建阶段 ----
FROM python:3.12-slim AS builder

WORKDIR /build

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

# 先复制依赖文件（利用 Docker 缓存层）
COPY requirements.txt .

# 安装 Python 依赖到虚拟环境
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
RUN pip install --no-cache-dir -r requirements.txt

# ---- 运行阶段 ----
FROM python:3.12-slim AS runtime

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

# 创建非 root 用户
RUN groupadd -r appuser && useradd -r -g appuser appuser

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

# 复制应用代码
WORKDIR /app
COPY app/ ./app/

# 切换到非 root 用户
USER appuser

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

# 暴露端口
EXPOSE 8000

# 启动命令（生产环境用 uvicorn，多 worker）
CMD ["uvicorn", "app.main:app", \
     "--host", "0.0.0.0", \
     "--port", "8000", \
     "--workers", "4", \
     "--loop", "uvloop", \
     "--http", "httptools", \
     "--log-level", "info", \
     "--access-log"]

# docker-compose.yml — 本地开发与生产部署
version: "3.9"

services:
  app:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: fastapi-app
    restart: unless-stopped
    ports:
      - "127.0.0.1:8000:8000"  # 仅绑定 localhost，Nginx 代理
    env_file:
      - .env
    environment:
      - RDS_HOST=${RDS_HOST}
      - RDS_PORT=5432
      - RDS_USER=${RDS_USER}
      - RDS_PASSWORD=${RDS_PASSWORD}
      - RDS_DATABASE=${RDS_DATABASE}
      - S3_BUCKET=${S3_BUCKET}
      - AWS_REGION=${AWS_REGION:-us-east-1}
    volumes:
      - app-logs:/app/logs
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 5s
      retries: 3
      start_period: 10s
    deploy:
      resources:
        limits:
          cpus: "2.0"
          memory: 1G
        reservations:
          cpus: "0.5"
          memory: 256M
    logging:
      driver: json-file
      options:
        max-size: "50m"
        max-file: "5"

volumes:
  app-logs:

#!/bin/bash
# scripts/deploy.sh — EC2 一键部署脚本

set -euo pipefail

STAGE="${1:-prod}"
COMMIT_SHA="${2:-$(git rev-parse --short HEAD)}"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)

echo "============================================"
echo "  部署 FastAPI 应用"
echo "  阶段: ${STAGE}"
echo "  版本: ${COMMIT_SHA}"
echo "  时间: ${TIMESTAMP}"
echo "============================================"

# 1. 拉取最新代码
echo "[1/6] 拉取最新代码..."
cd /opt/app
git fetch origin main
git checkout main
git pull origin main

# 2. 备份当前版本
echo "[2/6] 备份当前版本..."
if [ -f docker-compose.yml ]; then
    cp docker-compose.yml "docker-compose.yml.bak.${TIMESTAMP}"
fi

# 3. 构建新镜像
echo "[3/6] 构建 Docker 镜像..."
docker compose build --no-cache app
docker tag my-fastapi-app-app:latest "my-fastapi-app:${COMMIT_SHA}"

# 4. 停止旧容器
echo "[4/6] 停止旧容器..."
docker compose down --timeout 30

# 5. 启动新容器
echo "[5/6] 启动新容器..."
docker compose up -d

# 6. 健康检查
echo "[6/6] 健康检查..."
MAX_RETRIES=30
RETRY_COUNT=0
while [ $RETRY_COUNT -lt $MAX_RETRIES ]; do
    if curl -sf http://localhost:8000/health > /dev/null 2>&1; then
        echo "✓ 健康检查通过！"
        break
    fi
    RETRY_COUNT=$((RETRY_COUNT + 1))
    echo "  等待应用启动... ($RETRY_COUNT/$MAX_RETRIES)"
    sleep 2
done

if [ $RETRY_COUNT -eq $MAX_RETRIES ]; then
    echo "✗ 健康检查失败！回滚到上一版本..."
    docker compose down --timeout 10
    if [ -f "docker-compose.yml.bak.${TIMESTAMP}" ]; then
        cp "docker-compose.yml.bak.${TIMESTAMP}" docker-compose.yml
        docker compose up -d
    fi
    exit 1
fi

# 清理旧镜像
docker image prune -f --filter "until=168h"

echo ""
echo "============================================"
echo "  部署完成！"
echo "  版本: ${COMMIT_SHA}"
echo "  访问: https://example.com/health"
echo "============================================"

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3UploadAccess",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:PutObject",
        "s3:DeleteObject"
      ],
      "Resource": "arn:aws:s3:::my-app-uploads/*"
    },
    {
      "Sid": "S3ListBucket",
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::my-app-uploads",
      "Condition": {
        "StringLike": {
          "s3:prefix": ["uploads/*"]
        }
      }
    },
    {
      "Sid": "CloudWatchMetrics",
      "Effect": "Allow",
      "Action": [
        "cloudwatch:PutMetricData"
      ],
      "Resource": "*",
      "Condition": {
        "StringEquals": {
          "cloudwatch:namespace": "MyApp"
        }
      }
    },
    {
      "Sid": "KMSDecrypt",
      "Effect": "Allow",
      "Action": [
        "kms:Decrypt",
        "kms:DescribeKey"
      ],
      "Resource": "arn:aws:kms:us-east-1:123456789012:key/abcd1234"
    }
  ]
}

serverless-api/
├── src/
│   ├── handlers/
│   │   ├── __init__.py
│   │   ├── users.py      # 用户相关接口
│   │   ├── orders.py     # 订单相关接口
│   │   └── events.py     # 事件处理函数
│   ├── models/
│   │   ├── __init__.py
│   │   └── user.py
│   ├── services/
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── dynamodb.py
│   └── utils/
│       ├── __init__.py
│       ├── response.py    # 统一响应格式
│       └── logger.py
├── serverless.yml
├── requirements.txt
└── pyproject.toml

# src/utils/response.py — Lambda 统一响应格式
import json
from dataclasses import dataclass, field
from typing import Any


@dataclass
class LambdaResponse:
    """API Gateway Lambda 代理集成响应"""
    status_code: int = 200
    data: Any = None
    message: str = "success"
    errors: list[str] = field(default_factory=list)
    headers: dict[str, str] = field(default_factory=dict)

    def to_api_gateway_response(self) -> dict[str, Any]:
        """转换为 API Gateway 代理集成响应格式"""
        default_headers = {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Credentials": "true",
        }
        default_headers.update(self.headers)

        body = {
            "success": 200 <= self.status_code < 300,
            "message": self.message,
        }

        if self.data is not None:
            body["data"] = self.data
        if self.errors:
            body["errors"] = self.errors

        return {
            "statusCode": self.status_code,
            "headers": default_headers,
            "body": json.dumps(body, ensure_ascii=False, default=str),
        }


def success_response(data: Any = None, message: str = "success") -> dict[str, Any]:
    """成功响应快捷方法"""
    return LambdaResponse(data=data, message=message).to_api_gateway_response()


def error_response(
    status_code: int = 500,
    message: str = "Internal Server Error",
    errors: list[str] | None = None,
) -> dict[str, Any]:
    """错误响应快捷方法"""
    return LambdaResponse(
        status_code=status_code,
        message=message,
        errors=errors or [],
    ).to_api_gateway_response()

# src/utils/logger.py — Lambda 结构化日志
import json
import logging
import os
from datetime import datetime, timezone


class JSONFormatter(logging.Formatter):
    """JSON 格式日志（适合 CloudWatch Logs Insights 查询）"""

    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
            "request_id": getattr(record, "request_id", None),
            "stage": os.getenv("STAGE", "dev"),
        }

        if record.exc_info and record.exc_info[0] is not None:
            log_entry["exception"] = self.formatException(record.exc_info)

        # 附加自定义字段
        for attr in ["user_id", "duration_ms", "aws_request_id"]:
            value = getattr(record, attr, None)
            if value is not None:
                log_entry[attr] = value

        return json.dumps(log_entry, ensure_ascii=False)


def get_logger(name: str) -> logging.Logger:
    """获取配置好的 Logger"""
    logger = logging.getLogger(name)
    if not logger.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(JSONFormatter())
        logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    return logger

# src/services/dynamodb.py — DynamoDB 操作封装
import boto3
import os
from typing import Any
from botocore.exceptions import ClientError
from src.utils.logger import get_logger

logger = get_logger(__name__)


class DynamoDBService:
    """DynamoDB CRUD 操作封装"""

    def __init__(self, table_name: str | None = None) -> None:
        self.dynamodb = boto3.resource("dynamodb")
        self.table_name = table_name or os.getenv("DYNAMODB_TABLE", "app-data")
        self.table = self.dynamodb.Table(self.table_name)

    async def get_item(self, pk: str, sk: str) -> dict[str, Any] | None:
        """根据主键获取条目"""
        try:
            response = self.table.get_item(Key={"pk": pk, "sk": sk})
            return response.get("Item")
        except ClientError as e:
            logger.error(f"DynamoDB GetItem 失败: {e}")
            return None

    async def put_item(self, item: dict[str, Any]) -> bool:
        """写入条目"""
        try:
            self.table.put_item(Item=item)
            return True
        except ClientError as e:
            logger.error(f"DynamoDB PutItem 失败: {e}")
            return False

    async def query_by_pk(
        self,
        pk: str,
        sk_prefix: str | None = None,
        limit: int = 20,
    ) -> list[dict[str, Any]]:
        """根据分区键查询"""
        try:
            key_condition = boto3.dynamodb.conditions.Key("pk").eq(pk)
            if sk_prefix:
                key_condition &= boto3.dynamodb.conditions.Key("sk").begins_with(sk_prefix)

            response = self.table.query(
                KeyConditionExpression=key_condition,
                Limit=limit,
            )
            return response.get("Items", [])
        except ClientError as e:
            logger.error(f"DynamoDB Query 失败: {e}")
            return []

    async def delete_item(self, pk: str, sk: str) -> bool:
        """删除条目"""
        try:
            self.table.delete_item(Key={"pk": pk, "sk": sk})
            return True
        except ClientError as e:
            logger.error(f"DynamoDB DeleteItem 失败: {e}")
            return False

# src/handlers/users.py — 用户 CRUD 接口
import json
import uuid
from datetime import datetime, timezone

from src.services.dynamodb import DynamoDBService
from src.utils.response import success_response, error_response
from src.utils.logger import get_logger

logger = get_logger(__name__)
db = DynamoDBService()


def create_user(event: dict, context: dict) -> dict:
    """POST /api/v1/users — 创建用户"""
    try:
        body = json.loads(event.get("body", "{}"))
        name = body.get("name")
        email = body.get("email")

        if not name or not email:
            return error_response(
                status_code=400,
                message="缺少必填字段",
                errors=["name 和 email 为必填"],
            )

        user_id = str(uuid.uuid4())
        now = datetime.now(timezone.utc).isoformat()

        item = {
            "pk": f"USER#{user_id}",
            "sk": "PROFILE",
            "user_id": user_id,
            "name": name,
            "email": email,
            "created_at": now,
            "updated_at": now,
        }

        import asyncio
        success = asyncio.get_event_loop().run_until_complete(db.put_item(item))

        if not success:
            return error_response(status_code=500, message="创建用户失败")

        logger.info(f"用户创建成功: {user_id}", extra={"user_id": user_id})
        return success_response(data={"user_id": user_id, "name": name}, message="用户创建成功")

    except json.JSONDecodeError:
        return error_response(status_code=400, message="请求体格式错误")
    except Exception as e:
        logger.exception("创建用户异常")
        return error_response(status_code=500, message=str(e))


def get_user(event: dict, context: dict) -> dict:
    """GET /api/v1/users/{user_id} — 获取用户"""
    try:
        path_params = event.get("pathParameters") or {}
        user_id = path_params.get("user_id")

        if not user_id:
            return error_response(status_code=400, message="缺少 user_id")

        import asyncio
        item = asyncio.get_event_loop().run_until_complete(
            db.get_item(pk=f"USER#{user_id}", sk="PROFILE")
        )

        if not item:
            return error_response(status_code=404, message="用户不存在")

        return success_response(data=item)

    except Exception as e:
        logger.exception("获取用户异常")
        return error_response(status_code=500, message=str(e))


def list_users(event: dict, context: dict) -> dict:
    """GET /api/v1/users — 列出用户"""
    try:
        query = event.get("queryStringParameters") or {}
        limit = int(query.get("limit", "20"))

        import asyncio
        items = asyncio.get_event_loop().run_until_complete(
            db.query_by_pk(pk="USERS", sk_prefix="USER#", limit=limit)
        )

        return success_response(data={"items": items, "count": len(items)})

    except Exception as e:
        logger.exception("列出用户异常")
        return error_response(status_code=500, message=str(e))

#!/bin/bash
# scripts/serverless-deploy.sh — Serverless Framework 部署

set -euo pipefail

STAGE="${1:-dev}"

echo "=== 部署 Serverless API (${STAGE}) ==="

# 1. 安装 Serverless Framework
npm install -g serverless

# 2. 安装 Python 插件
npm install --save-dev serverless-python-requirements

# 3. 部署
echo "[1/3] 部署到 AWS Lambda (${STAGE})..."
serverless deploy --stage "${STAGE}"

# 4. 查看部署信息
echo "[2/3] 部署信息:"
serverless info --stage "${STAGE}"

# 5. 验证
API_URL=$(serverless info --stage "${STAGE}" | grep "endpoint:" | awk '{print $2}')
echo "[3/3] 验证部署..."
if curl -sf "${API_URL}/api/v1/users" > /dev/null; then
    echo "✓ 部署验证通过"
else
    echo "✗ 部署验证失败，请检查日志"
    serverless logs --stage "${STAGE}" --function api
fi

echo ""
echo "=== 部署完成 ==="
echo "API 地址: ${API_URL}"

# 常用 Serverless 命令

# 部署
serverless deploy --stage prod

# 查看日志
serverless logs --function api --stage prod --tail

# 调用函数
serverless invoke --function api --stage prod --data '{"httpMethod":"GET","path":"/api/v1/users"}'

# 回滚（回退到上一个部署版本）
serverless rollback --stage prod

# 查看指标
serverless metrics --stage prod

# 删除所有资源
serverless remove --stage prod

# ❌ 问题：每次冷启动都要重新初始化全局资源

import boto3

def handler(event, context):
    # 每次调用都创建新的客户端（冷启动 + 热启动都创建）
    dynamodb = boto3.resource("dynamodb")
    table = dynamodb.Table("my-table")
    s3 = boto3.client("s3")
    # ... 业务逻辑


# ✅ 解决：在全局作用域初始化（只冷启动时执行一次，热启动复用）

import boto3

# 全局初始化（容器复用时不会重新执行）
dynamodb = boto3.resource("dynamodb")
TABLE = dynamodb.Table("my-table")
S3_CLIENT = boto3.client("s3")


def handler(event, context):
    # 热启动时直接使用已初始化的客户端
    response = TABLE.get_item(Key={"pk": "123"})
    # ... 业务逻辑

# ❌ 问题：导入沉重的库导致冷启动慢

# handler.py — 包含所有依赖，冷启动加载所有模块
import pandas        # ~100ms
import numpy         # ~80ms
import sqlalchemy    # ~150ms
from my_app import everything  # 全量导入


# ✅ 解决：按需导入 + Lambda Layer 分离

# handler.py — 只导入必要的轻量模块
import json
import boto3

def handler(event, context):
    path = event.get("path", "")

    if path.startswith("/api/analytics"):
        # 仅在需要时导入重型依赖
        import pandas as pd  # noqa: F811
        return _handle_analytics(event)

    # 常规接口不需要 pandas
    return _handle_default(event)