Alembic 迁移和启动装配

前面已经有配置、数据库、路由、异常、JWT。

最后要把项目装配完整，并用 Alembic 管理表结构。

一、初始化 Alembic

在项目根目录执行：

uv run alembic init alembic

生成：

alembic/
├── env.py
├── script.py.mako
└── versions/
alembic.ini

二、让 Alembic 使用项目配置

默认 alembic.ini 会维护一份数据库地址。企业项目建议统一从 .env 的 DATABASE_URL 读取。

alembic/env.py 关键修改：

from logging.config import fileConfig

from alembic import context
from sqlalchemy import engine_from_config, pool

from app.core.config import settings
from app.db.base import Base

# 导入所有模型，保证 autogenerate 能读取完整 metadata。
import app.models  # noqa: F401

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# URL 里可能有 %，写入 ConfigParser 前需要转义。
config.set_main_option("sqlalchemy.url", settings.database_url.replace("%", "%%"))

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True,
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,
        )

        with context.begin_transaction():
            context.run_migrations()


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

关键点：

配置	说明
`settings.database_url`	Alembic 和应用使用同一个数据库地址
`import app.models`	自动迁移必须导入所有模型
`target_metadata = Base.metadata`	让 Alembic 对比 ORM 模型
`compare_type=True`	字段类型变化也能被检测

三、导入所有模型

app/models/__init__.py：

from app.models.user import User

__all__ = ["User"]

以后新增模型，例如 Post，也要在这里导入：

from app.models.post import Post

否则 Alembic 自动生成迁移时可能看不到新模型。

四、生成和执行迁移

首次建表：

uv run alembic revision --autogenerate -m "create users table"
uv run alembic upgrade head

以后修改表结构：

# 1. 修改 app/models/*.py
# 2. 生成迁移草稿
uv run alembic revision --autogenerate -m "add user nickname"

# 3. 人工检查 alembic/versions/ 下的迁移文件
# 4. 执行迁移
uv run alembic upgrade head

常用命令：

uv run alembic current
uv run alembic history
uv run alembic downgrade -1

如果数据库表已经手动存在，并且结构和当前模型一致，可以先标记版本：

uv run alembic stamp head

不要没检查就直接对生产数据库执行自动生成的迁移文件。

五、应用工厂装配

app/main.py：

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.api.router import api_router
from app.core.config import settings
from app.core.exceptions import register_exception_handlers
from app.core.response import success
from app.db.session import check_database_connection
from app.middlewares.auth_middleware import AuthMiddleware
from app.schemas.response_schema import ApiResponse


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # 本地开发时快速发现数据库配置问题。
    check_database_connection()
    yield


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.app_name,
        description="企业级 FastAPI 示例项目接口文档",
        version="0.1.0",
        debug=settings.debug,
        lifespan=lifespan,
        openapi_tags=[
            {"name": "系统", "description": "健康检查、应用信息"},
            {"name": "鉴权", "description": "注册、登录、Token"},
            {"name": "用户", "description": "用户管理接口"},
        ],
    )

    register_exception_handlers(app)

    # 默认接口需要登录，白名单接口由中间件内部放行。
    app.add_middleware(AuthMiddleware)

    app.include_router(api_router, prefix="/api")

    @app.get("/", response_model=ApiResponse[dict], tags=["系统"])
    def index():
        return success(data={"name": settings.app_name, "env": settings.app_env})

    return app


app = create_app()

注册顺序建议：

创建 FastAPI。
配置 OpenAPI 和文档。
注册异常处理器。
注册中间件。
注册路由。
定义根路径或健康检查。

六、顶层路由聚合

app/api/router.py：

from fastapi import APIRouter

from app.api.routes import auth_routes, system_routes, user_routes

api_router = APIRouter()
api_router.include_router(system_routes.router)
api_router.include_router(auth_routes.router)
api_router.include_router(user_routes.router)

app/api/routes/system_routes.py：

from fastapi import APIRouter

from app.controllers.health_controller import health_controller

router = APIRouter(tags=["系统"])
router.add_api_route("/health", health_controller, methods=["GET"], summary="健康检查")

app/controllers/health_controller.py：

from app.core.response import success


def health_controller():
    return success(data={"status": "ok"})

健康检查建议放到 JWT 白名单里：

"/api/health"

七、本地启动入口

可以直接启动：

uv run fastapi dev app/main.py

也可以创建根目录 main.py，做更友好的本地启动提示：

import uvicorn

from app.core.config import settings


def main():
    # 本地开发入口，生产环境建议使用进程管理工具启动。
    uvicorn.run(
        "app.main:app",
        host=settings.app_host,
        port=settings.app_port,
        reload=True,
    )


if __name__ == "__main__":
    main()

运行：

uv run main.py

八、从零搭建步骤总览

uv init my-fastapi
安装 FastAPI、SQLAlchemy、Alembic、PyJWT、Pydantic Settings。
创建 app/ 分层目录。
写 .env.example 和 .gitignore。
写 core/config.py。
写 db/base.py、db/session.py。
写 core/response.py、schemas/response_schema.py。
写 core/exceptions.py 并在 create_app() 注册。
写 models/user.py。
初始化 Alembic，生成并执行迁移。
写 core/security.py。
写登录注册的 schema、service、controller、routes。
写 AuthMiddleware 和 get_current_user。
写业务模块，例如用户列表。
启动服务，先调用登录接口，再用 Token 调试受保护接口。

九、上线前检查

检查项	要求
`.env`	不提交真实 `.env`
`JWT_SECRET_KEY`	至少 32 字节，生产环境单独配置
数据库迁移	迁移文件人工检查后再执行
响应格式	成功和失败都保持统一
认证白名单	只放真正公开的接口
密码	只保存哈希，不保存明文
日志	不打印数据库密码、Token、明文密码
文档	Swagger 只用于开发和内部调试

到这里，一个标准 MVC 分层的 FastAPI 后端项目就搭起来了。

#Alembic 迁移和启动装配

#一、初始化 Alembic

#二、让 Alembic 使用项目配置

#三、导入所有模型

#四、生成和执行迁移

#五、应用工厂装配

#六、顶层路由聚合

#七、本地启动入口

#八、从零搭建步骤总览

#九、上线前检查