FastAPI 项目实践

最近用 FastAPI 写了一个新项目，这是我第一次在实际项目里完整使用 FastAPI。项目的技术栈主要是 FastAPI + PostgreSQL + SQLAlchemy Async ORM + Alembic，接口层使用异步请求，数据库访问也采用异步方式。

这篇文章主要记录我从零搭建项目的过程，包括项目结构、数据库连接、异步 Session 管理、ORM 模型定义、Alembic 迁移配置，以及开发过程中踩到的一些坑。

FastAPI 本身并不强制绑定某一种数据库或 ORM，它更像是一个轻量、高性能的 Web 框架。数据库层可以根据项目需要选择 SQLAlchemy、SQLModel、Tortoise ORM 或其他工具。这里我选择的是 SQLAlchemy，因为它生态成熟，并且在 2.x 版本中对异步能力支持得比较完整。FastAPI 官方文档也说明，FastAPI 可以配合任意关系型数据库使用，SQLAlchemy / SQLModel 都是常见选择。 FastAPI SQLAlchemy

---

一、项目技术栈

这个项目的后端主要使用以下技术：

技术	作用
FastAPI	Web 框架，负责接口定义、请求处理、依赖注入
PostgreSQL	关系型数据库
SQLAlchemy 2.x	ORM，负责模型映射和数据库操作
asyncpg	PostgreSQL 异步驱动
Alembic	数据库迁移工具
Pydantic	请求参数和响应数据校验
Uvicorn	ASGI 服务启动器

整体思路是：FastAPI 负责 API 层，SQLAlchemy 负责数据库 ORM，asyncpg 负责异步连接 PostgreSQL，Alembic 负责维护数据库表结构的版本变更。

---

二、初始化项目

首先创建项目目录：

mkdir fastapi-demo
cd fastapi-demo

创建虚拟环境：

python -m venv .venv

激活虚拟环境：

# macOS / Linux
source .venv/bin/activate

# Windows
.venv\Scripts\activate

安装依赖：

pip install fastapi uvicorn sqlalchemy asyncpg alembic pydantic-settings

如果后续需要做密码加密、JWT 登录、环境变量管理等功能，也可以再安装：

pip install passlib[bcrypt] python-jose python-multipart

---

三、推荐的项目结构

我这次采用了一个比较清晰的分层结构：

fastapi-demo/
├── alembic/
│   ├── versions/
│   └── env.py
├── app/
│   ├── api/
│   │   └── user.py
│   ├── core/
│   │   └── config.py
│   ├── db/
│   │   ├── base.py
│   │   └── session.py
│   ├── models/
│   │   └── user.py
│   ├── schemas/
│   │   └── user.py
│   ├── services/
│   │   └── user.py
│   └── main.py
├── alembic.ini
├── .env
└── requirements.txt

这个结构的好处是职责比较清楚：

api 层只负责路由和接口参数；schemas 层负责请求和响应数据结构；models 层负责数据库模型；services 层负责业务逻辑；db 层负责数据库连接和 Session 管理；core 层负责配置项。

刚开始写 FastAPI 时，很容易把所有代码都堆到 main.py 里。小 demo 这样写没问题，但项目稍微复杂一点，就会变得很难维护。所以我更推荐一开始就做基础分层。

---

四、配置环境变量

在项目根目录创建 .env 文件：

DATABASE_URL=postgresql+asyncpg://postgres:password@localhost:5432/fastapi_demo

这里要注意，异步连接 PostgreSQL 时，数据库 URL 需要使用：

postgresql+asyncpg://

而不是传统同步连接里的：

postgresql://

然后创建配置文件 app/core/config.py：

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    DATABASE_URL: str

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8"
    )


settings = Settings()

这样我们就可以通过 settings.DATABASE_URL 统一读取数据库连接地址。

---

五、创建异步数据库连接

接下来创建 app/db/session.py：

from collections.abc import AsyncGenerator

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.core.config import settings


engine = create_async_engine(
    settings.DATABASE_URL,
    echo=True,
    pool_pre_ping=True,
)


AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        yield session

这里有几个关键点。

create_async_engine 用来创建异步数据库引擎。async_sessionmaker 用来创建异步 Session 工厂。get_db 是 FastAPI 的依赖函数，在接口里可以通过 Depends(get_db) 注入数据库 Session。

expire_on_commit=False 也比较重要。默认情况下，SQLAlchemy 在 commit 后可能会让对象属性过期，后续再次访问属性时会触发数据库加载。在异步场景里，如果处理不好，可能会遇到额外的懒加载问题。对于大多数 API 项目，把它设置为 False 会更直观一些。

SQLAlchemy 的异步 ORM 使用 AsyncSession、create_async_engine 和 async_sessionmaker 来完成异步数据库访问，这也是 FastAPI 项目里比较常见的组合方式。 SQLAlchemy

---

六、定义 ORM 基类

创建 app/db/base.py：

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

所有 ORM 模型都会继承这个 Base。Alembic 后面也会通过这个 Base.metadata 来识别当前项目里的表结构。

---

七、创建 User 模型

创建 app/models/user.py：

from datetime import datetime

from sqlalchemy import Boolean, DateTime, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.db.base import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True, nullable=False)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True, nullable=False)
    hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        nullable=False,
    )

这里使用的是 SQLAlchemy 2.x 推荐的类型写法：Mapped + mapped_column。

相比老版本写法，这种方式类型提示更友好，也更适合现代 Python 项目。比如编辑器可以更好地推断字段类型，后续写查询逻辑时体验会更好。

---

八、创建 Pydantic Schema

数据库模型不建议直接作为接口响应返回。通常我们会单独定义 Pydantic Schema，用来描述请求体和响应体。

创建 app/schemas/user.py：

from datetime import datetime

from pydantic import BaseModel, EmailStr


class UserCreate(BaseModel):
    email: EmailStr
    username: str
    password: str


class UserRead(BaseModel):
    id: int
    email: EmailStr
    username: str
    is_active: bool
    created_at: datetime

    model_config = {
        "from_attributes": True
    }

UserCreate 用于创建用户时接收请求参数。

UserRead 用于返回用户信息。这里没有返回 hashed_password，因为密码哈希不应该暴露给前端。

from_attributes=True 是 Pydantic v2 中从 ORM 对象读取数据时常用的配置。

---

九、编写 Service 层

创建 app/services/user.py：

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.user import User
from app.schemas.user import UserCreate


async def get_user_by_email(db: AsyncSession, email: str) -> User | None:
    stmt = select(User).where(User.email == email)
    result = await db.execute(stmt)
    return result.scalar_one_or_none()


async def create_user(db: AsyncSession, user_in: UserCreate) -> User:
    fake_hashed_password = f"hashed_{user_in.password}"

    user = User(
        email=user_in.email,
        username=user_in.username,
        hashed_password=fake_hashed_password,
    )

    db.add(user)
    await db.commit()
    await db.refresh(user)

    return user

这里我把数据库操作放到了 services 层，而不是直接写在路由函数里。

这样做的好处是，路由层会更薄，只处理 HTTP 相关逻辑；业务逻辑和数据库逻辑放在 service 层，后面也更容易写测试。

需要注意的是，异步 SQLAlchemy 查询时要使用：

result = await db.execute(stmt)

然后再通过：

result.scalar_one_or_none()

拿到 ORM 对象。

---

十、编写 API 路由

创建 app/api/user.py：

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import get_db
from app.schemas.user import UserCreate, UserRead
from app.services.user import create_user, get_user_by_email


router = APIRouter(prefix="/users", tags=["users"])


@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user_api(
    user_in: UserCreate,
    db: AsyncSession = Depends(get_db),
):
    existing_user = await get_user_by_email(db, user_in.email)

    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered",
        )

    return await create_user(db, user_in)

然后在 app/main.py 中注册路由：

from fastapi import FastAPI

from app.api.user import router as user_router


app = FastAPI(
    title="FastAPI Demo",
    version="0.1.0",
)


app.include_router(user_router)


@app.get("/health")
async def health_check():
    return {"status": "ok"}

启动项目：

uvicorn app.main:app --reload

打开浏览器访问：

http://127.0.0.1:8000/docs

就可以看到 FastAPI 自动生成的 Swagger 文档。

这也是 FastAPI 开发体验很好的地方：只要写好类型提示、请求模型和响应模型，接口文档基本可以自动生成。

---

十一、配置 Alembic 数据库迁移

在项目根目录初始化 Alembic：

alembic init alembic

执行后会生成：

alembic/
├── versions/
└── env.py

alembic.ini

接下来要做两件事：

第一，把 Alembic 的数据库地址配置成项目里的数据库地址。

第二，让 Alembic 能识别 SQLAlchemy 的模型元数据。

---

十二、修改 alembic.ini

打开 alembic.ini，找到：

sqlalchemy.url = driver://user:pass@localhost/dbname

这里可以先留空，或者写一个占位值：

sqlalchemy.url =

因为我们会在 env.py 里从项目配置中读取真实数据库地址。

---

十三、修改 Alembic env.py 支持异步迁移

重点来了。

如果项目使用的是：

postgresql+asyncpg://

那么 Alembic 的迁移配置也需要适配异步数据库连接。

修改 alembic/env.py：

import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context

from app.core.config import settings
from app.db.base import Base
from app.models import user  # 确保模型被导入，否则 Alembic 可能识别不到


config = context.config

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


config.set_main_option("sqlalchemy.url", settings.DATABASE_URL)

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"},
    )

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


def do_run_migrations(connection: Connection) -> None:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
    )

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


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

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())


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

这里最关键的地方是：

await connection.run_sync(do_run_migrations)

Alembic 的迁移上下文本身仍然是同步风格的，但是数据库连接是异步的，所以需要通过 run_sync 把同步迁移逻辑挂到异步连接上执行。

Alembic 官方提供了异步模板，核心思路就是在 env.py 中使用异步 engine，并通过 connection.run_sync() 执行迁移逻辑。Alembic 本身是 SQLAlchemy 生态里的数据库迁移工具，通常用来管理表结构的版本变更。 Alembic GitHub

---

十四、让 Alembic 能识别所有模型

上面的 env.py 里有一行：

from app.models import user

这行看起来好像没有被直接使用，但它很重要。

Alembic 自动生成迁移文件时，会读取：

target_metadata = Base.metadata

但是只有模型类被 Python 导入之后，它们才会注册到 Base.metadata 上。

所以，如果忘记导入模型，执行：

alembic revision --autogenerate -m "create users table"

可能会发现生成的迁移文件是空的。

更好的做法是，在 app/models/__init__.py 中统一导入所有模型：

from app.models.user import User

然后在 alembic/env.py 中写：

import app.models

这样后面模型越来越多时，不需要在 env.py 里一个个导入。

---

十五、生成并执行迁移

生成迁移文件：

alembic revision --autogenerate -m "create users table"

如果配置正确，Alembic 会在 alembic/versions/ 目录下生成一个迁移文件，里面大概会有类似这样的内容：

def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("email", sa.String(length=255), nullable=False),
        sa.Column("username", sa.String(length=50), nullable=False),
        sa.Column("hashed_password", sa.String(length=255), nullable=False),
        sa.Column("is_active", sa.Boolean(), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_users_id"), "users", ["id"], unique=False)
    op.create_index(op.f("ix_users_email"), "users", ["email"], unique=True)
    op.create_index(op.f("ix_users_username"), "users", ["username"], unique=True)


def downgrade() -> None:
    op.drop_index(op.f("ix_users_username"), table_name="users")
    op.drop_index(op.f("ix_users_email"), table_name="users")
    op.drop_index(op.f("ix_users_id"), table_name="users")
    op.drop_table("users")

执行迁移：

alembic upgrade head

查看当前迁移版本：

alembic current

回滚一个版本：

alembic downgrade -1

---

十六、测试创建用户接口

启动服务：

uvicorn app.main:app --reload

访问：

http://127.0.0.1:8000/docs

调用：

POST /users

请求体：

{
  "email": "test@example.com",
  "username": "testuser",
  "password": "123456"
}

返回结果类似：

{
  "id": 1,
  "email": "test@example.com",
  "username": "testuser",
  "is_active": true,
  "created_at": "2026-06-20T08:00:00Z"
}

这说明 FastAPI、PostgreSQL、SQLAlchemy 异步 Session 和 Alembic 迁移已经基本跑通了。

---

十七、开发过程中踩到的坑

1. Alembic 不是 ORM

一开始我也容易把 Alembic 和 ORM 混在一起理解。

后来才理清楚：

SQLAlchemy 是 ORM，负责把 Python 类映射成数据库表，并提供查询能力。

Alembic 是迁移工具，负责记录数据库表结构的变化，比如创建表、添加字段、删除索引等。

两者经常一起使用，但职责完全不同。

---

2. PostgreSQL 异步连接要使用 asyncpg

如果项目是异步数据库操作，数据库地址需要写成：

postgresql+asyncpg://user:password@host:port/dbname

如果写成普通的：

postgresql://user:password@host:port/dbname

可能会导致异步 engine 无法正常工作。

---

3. 忘记导入模型会导致迁移文件为空

Alembic 的 --autogenerate 依赖 Base.metadata。

如果模型没有被导入，Base.metadata 里面就没有对应的表信息，迁移文件自然可能是空的。

所以一定要保证在 alembic/env.py 中导入所有模型。

---

4. async 函数里不要忘记 await

使用异步 SQLAlchemy 时，很多操作都需要 await：

result = await db.execute(stmt)
await db.commit()
await db.refresh(user)

如果忘记 await，可能不会立刻报出特别直观的错误，但程序行为会不符合预期。

---

5. 不要在路由里堆太多业务逻辑

刚开始写 FastAPI，很容易把查询、校验、创建、异常处理都写在一个路由函数里。

短期看起来很快，但后面接口一多就会变乱。

我现在更倾向于：

路由层负责 HTTP。

Service 层负责业务逻辑。

Model 层负责数据库结构。

Schema 层负责输入输出数据结构。

这样项目会更清晰，也更方便测试。

---

十八、一个完整请求的执行流程

以创建用户接口为例，请求流程大概是这样的：

sequenceDiagram
    participant Client as Client
    participant API as FastAPI Router
    participant Service as User Service
    participant DBSession as AsyncSession
    participant PostgreSQL as PostgreSQL

    Client->>API: POST /users
    API->>API: 校验 UserCreate
    API->>DBSession: Depends(get_db)
    API->>Service: get_user_by_email()
    Service->>PostgreSQL: SELECT users WHERE email = ?
    PostgreSQL-->>Service: 查询结果
    Service-->>API: 用户不存在
    API->>Service: create_user()
    Service->>DBSession: db.add(user)
    Service->>PostgreSQL: INSERT INTO users
    Service->>DBSession: commit + refresh
    Service-->>API: 返回 User 对象
    API-->>Client: 返回 UserRead

这个流程也是我理解 FastAPI 依赖注入和异步数据库操作的关键。

FastAPI 的 Depends(get_db) 会在请求进入路由时创建数据库 Session，请求结束后自动退出上下文。Service 层拿到这个 Session 之后，就可以执行数据库查询和写入。

---

十九、我的项目实践建议

如果是第一次用 FastAPI，我建议不要一上来就引入太多复杂封装。

可以先把核心链路跑通：

FastAPI 路由
↓
Pydantic Schema
↓
SQLAlchemy Model
↓
AsyncSession
↓
PostgreSQL
↓
Alembic Migration

等这个流程跑通之后，再考虑加入认证、权限、缓存、日志、测试、Docker、CI/CD 等内容。

我这次最大的感受是，FastAPI 本身并不难，真正需要花时间理解的是它和数据库层的组合方式。尤其是异步 SQLAlchemy 和 Alembic 的配置，刚开始看起来会有点绕，但只要理解了 AsyncSession、Base.metadata 和 env.py 的关系，后面就会清晰很多。

---

二十、后续可以继续优化的方向

目前这个项目只是完成了基础架构，后续还可以继续优化：

1. 使用真实的密码哈希，比如 passlib[bcrypt]。
2. 加入 JWT 登录认证。
3. 使用 Docker Compose 管理 PostgreSQL。
4. 增加 pytest 异步测试。
5. 增加统一异常处理。
6. 增加日志中间件。
7. 把配置分成开发环境、测试环境和生产环境。
8. 增加数据库连接池参数配置。
9. 对 Service 层做更细的业务拆分。
10. 在 CI/CD 中自动执行测试和迁移检查。

FastAPI 的优势在于开发效率高、类型提示友好、自动文档完善，再配合 PostgreSQL、SQLAlchemy 和 Alembic，已经可以支撑一个比较规范的后端 API 项目。