《Python编程实战》18.3 上线、复盘与后续演进

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="TASKFLOW_", env_file=".env")
    database_url: str = "sqlite+aiosqlite:///./taskflow.db"
    secret_key: str = "dev-only-secret-change-me"
    access_token_ttl: int = 3600
    cache_ttl: int = 60

from pydantic import model_validator

class Settings(BaseSettings):
    app_env: str = "development"
    secret_key: str = "dev-only-secret-change-me"

    @model_validator(mode="after")
    def guard_production(self) -> "Settings":
        if self.app_env == "production" and self.secret_key == "dev-only-secret-change-me":
            raise ValueError("生产环境必须显式设置 TASKFLOW_SECRET_KEY")
        return self

FROM python:3.14-slim AS base
WORKDIR /app
COPY pyproject.toml ./
RUN pip install --no-cache-dir .
COPY app ./app
RUN useradd --create-home appuser
USER appuser
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

$ alembic upgrade head
INFO  [alembic.runtime.migration] Running upgrade  -> 9ae51e968831, init tickets schema
INFO  [alembic.runtime.migration] Running upgrade 9ae51e968831 -> c021935f2bed, add ticket due_at
$ alembic current
c021935f2bed (head)

$ alembic downgrade -1
INFO  [alembic.runtime.migration] Running downgrade c021935f2bed -> 9ae51e968831, add ticket due_at
$ alembic current
9ae51e968831
tickets 列: ['id', 'title', 'body', 'status', 'priority', 'owner_id', 'created_at']

INFO:     Started server process [58074]
INFO:     Waiting for application startup.
2026-10-09 12:11:08,394 INFO database ready
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8019 (Press CTRL+C to quit)
INFO:     Shutting down
INFO:     Waiting for application shutdown.
2026-10-09 12:11:08,604 INFO shutdown complete
INFO:     Application shutdown complete.

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    log.info("database ready")
    yield
    await tasks.worker()      # 关闭前把队列里的任务处理完
    await engine.dispose()    # 释放连接池
    log.info("shutdown complete")

@app.middleware("http")
async def add_timing(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    response.headers["X-Process-Time-ms"] = f"{(time.perf_counter() - start) * 1000:.2f}"
    return response

from prometheus_client import CONTENT_TYPE_LATEST, Histogram, generate_latest

REQ_LATENCY = Histogram("http_request_seconds", "请求耗时", ["method", "status"])

@app.middleware("http")
async def observe(request: Request, call_next):
    response = await call_next(request)
    REQ_LATENCY.labels(request.method, str(response.status_code)).observe(
        response.headers.get("X-Process-Time-ms", 0)  # 简化示意
    )
    return response

@app.get("/metrics", include_in_schema=False)
async def metrics():
    return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)

## 迭代复盘 <版本号>
### 做对了什么（保持）
  - 契约先行：接口先定死，前后端不互相等
  - 租户隔离收敛到一个函数，越权返回 404
  - 后台任务一次做齐幂等/重试/死信
### 踩了什么坑（改进）
  - httpx 默认走系统代理，本地联调被拦 -> 测试脚本统一 trust_env=False
  - create_all 与 alembic 混用导致版本失联 -> 生产禁用 create_all
### 下一步（行动项）
  - 补 /metrics 与 OpenTelemetry 埋点
  - 给登录接口加限流（scrypt 每次约 36 ms，易被刷）