企业官网建设流程全解析

数据库实战：FastAPI + SQLAlchemy 2.0 + Alembic 从零搭建，踩坑实录

前言在现代 Web 开发中，数据库操作是核心环节之一。FastAPI 作为高性能异步框架，与 SQLAlchemy 2.0 的组合堪称黄金搭档，而 Alembic 则能帮助我们优雅地管理数据库迁移。本文将带你从零搭建一套完整的数据库操作环境，并分享我在这条路上踩过的坑。## 环境搭建与依赖配置### 核心依赖版本- Python 3.10± FastAPI 0.104± SQLAlchemy 2.0.23± Alembic 1.13.0± asyncpg (异步 PostgreSQL 驱动)### 踩坑点 1：SQLAlchemy 2.0 的异步模式SQLAlchemy 2.0 引入全新的异步 API，与旧版本不兼容。新手容易混淆create_engine和create_async_engine。python# database.py - 数据库配置核心文件from sqlalchemy.ext.asyncio import create_async_engine, AsyncSessionfrom sqlalchemy.orm import sessionmaker, DeclarativeBase# 注意：必须使用 async 前缀的引擎DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/testdb"engine = create_async_engine(DATABASE_URL, echo=True)# AsyncSession 是异步会话工厂async_session = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False # 避免事务提交后对象过期)class Base(DeclarativeBase): passasync def get_db() -> AsyncSession: """依赖注入：获取数据库会话""" async with async_session() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close()## 模型定义与数据操作### 踩坑点 2：异步操作中的上下文管理忘记在异步函数中使用async with或await会导致连接泄漏。以下是一个完整的用户模型示例：python# models.py - 定义数据库模型from sqlalchemy import Column, Integer, String, DateTime, funcfrom sqlalchemy.orm import relationshipfrom database import Baseclass User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String(50), unique=True, nullable=False, index=True) email = Column(String(120), unique=True, nullable=False) created_at = Column(DateTime(timezone=True), server_default=func.now()) # 关系定义（可选） # posts = relationship("Post", back_populates="author") def __repr__(self): return f"<User(id={self.id}, username='{self.username}')>"# schemas.py - Pydantic 模型（用于 API 请求验证）from pydantic import BaseModel, EmailStrfrom typing import Optionalfrom datetime import datetimeclass UserCreate(BaseModel): username: str email: EmailStrclass UserResponse(BaseModel): id: int username: str email: str created_at: datetime class Config: from_attributes = True # 支持 ORM 模式## FastAPI 路由实现### 踩坑点 3：异步 CRUD 操作的正确写法很多新手会忘记在数据库操作前加await，或者错误地使用同步查询。以下是一个完整的用户管理 API：python# main.py - FastAPI 应用主文件from fastapi import FastAPI, Depends, HTTPException, statusfrom sqlalchemy.ext.asyncio import AsyncSessionfrom sqlalchemy import selectfrom typing import Listfrom database import get_dbfrom models import Userfrom schemas import UserCreate, UserResponseapp = FastAPI(title="User Management API")@app.post("/users/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)async def create_user(user_data: UserCreate, db: AsyncSession = Depends(get_db)): """ 创建新用户 关键点：使用 await 执行异步查询 """ # 检查用户名是否已存在 result = await db.execute( select(User).where(User.username == user_data.username) ) existing_user = result.scalar_one_or_none() if existing_user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already exists" ) # 创建新用户 new_user = User( username=user_data.username, email=user_data.email ) db.add(new_user) await db.flush() # 确保获取到 id await db.refresh(new_user) # 刷新对象状态 return new_user@app.get("/users/{user_id}", response_model=UserResponse)async def get_user(user_id: int, db: AsyncSession = Depends(get_db)): """获取单个用户""" result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if not user: raise HTTPException(status_code=404, detail="User not found") return user@app.get("/users/", response_model=List[UserResponse])async def list_users(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)): """分页获取用户列表""" result = await db.execute( select(User).offset(skip).limit(limit) ) users = result.scalars().all() return users## Alembic 数据库迁移配置### 踩坑点 4：异步引擎与 Alembic 的兼容性Alembic 默认使用同步引擎，需要手动配置异步支持。以下是一个完整的迁移配置：python# alembic/env.py - 关键配置部分from logging.config import fileConfigfrom sqlalchemy import engine_from_configfrom sqlalchemy import poolfrom alembic import contextimport asynciofrom sqlalchemy.ext.asyncio import create_async_engine# 导入你的 Base 和模型from database import Basefrom models import User # 确保模型被导入config = context.configfileConfig(config.config_file_name)target_metadata = Base.metadatadef run_migrations_offline(): """离线迁移模式""" 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): """执行迁移的核心函数""" context.configure( connection=connection, target_metadata=target_metadata ) with context.begin_transaction(): context.run_migrations()async def run_async_migrations(): """异步迁移入口（关键修复）""" connectable = create_async_engine( config.get_main_option("sqlalchemy.url"), poolclass=pool.NullPool, ) async with connectable.connect() as connection: await connection.run_sync(do_run_migrations) await connectable.dispose()def run_migrations_online(): """在线迁移模式（使用异步）""" asyncio.run(run_async_migrations())if context.is_offline_mode(): run_migrations_offline()else: run_migrations_online()### 迁移命令示例bash# 初始化 Alembicalembic init alembic# 生成迁移脚本alembic revision --autogenerate -m "create_users_table"# 执行迁移alembic upgrade head# 回滚alembic downgrade -1## 踩坑总结与最佳实践### 常见问题速查表| 问题现象 | 原因 | 解决方案 ||---------|------|----------||AsyncSession对象未关闭 | 忘记使用async with| 始终使用上下文管理器 || 事务提交后对象属性为 None |expire_on_commit=True| 设置为 False 或手动 refresh || 异步查询返回空结果 | 忘记await| 检查所有数据库操作前加 await || Alembic 迁移报连接错误 | 使用了同步引擎 | 配置异步引擎适配器 || 外键约束错误 | 模型定义顺序问题 | 确保关联模型先定义 |### 性能优化建议1.连接池配置：根据并发量调整pool_size和max_overflow2.懒加载 vs 立即加载：使用selectinload避免 N+1 查询3.批量操作：使用bulk_insert_mappings提高插入性能4.索引优化：为常用查询字段添加索引### 最终可运行示例将以上代码整合后，运行以下命令即可体验完整功能：bash# 1. 创建虚拟环境python -m venv venvsource venv/bin/activate# 2. 安装依赖pip install fastapi uvicorn sqlalchemy alembic asyncpg pydantic# 3. 启动服务uvicorn main:app --reload# 4. 访问 API 文档# http://localhost:8000/docs## 总结通过本文，我们从零搭建了一套完整的 FastAPI + SQLAlchemy 2.0 + Alembic 技术栈，涵盖了环境配置、模型定义、异步 CRUD 操作、数据库迁移等关键环节。踩坑记录揭示了异步编程中常见的陷阱，包括上下文管理、事务状态、引擎类型选择等。记住：异步不是魔法，而是需要严谨对待的编程范式。在实际项目中，建议结合单元测试和集成测试来验证数据库操作的正确性，同时关注 SQLAlchemy 2.0 的官方文档更新，因为异步 API 仍在持续优化中。掌握这套技术栈，你将能高效构建健壮的后端系统。