刚开始学习FastAPI ORM时我以为“建表”就是定义几个字段。真正写起来才发现代码中同时出现了异步驱动、Engine、连接池、模型类、metadata、session、生命周期、uvicorn。任何一环没有接好最终都可能出现同一种现象代码看起来运行了数据库里却没有表。经过实际排错我把整个过程整理成了一条链路启动uvicorn-fastapi进入生命周期-engine根据模型创建表-metadata根据模型创建表-session进入路由-执行查询和增删改查只要判断问题发生在哪一环orm就不会显的那么混乱。一、ORM中几个最重要的对象不用ORM时我们直接写SQLSELECT * FROM book WHERE id 1;使用SQLALlchemy ORM后可以写成stmt select(Book).where(Book.id 1)可以把ORM中的几个对象理解为engine管理数据库连接和连接池的引擎model描述python类与数据库表的关系session负责一次具体的数据库操作和事务select描述需要查询什么这里定义模型并不等于执行建表写出select也不等于已经访问数据库。模型和查询语句都只是描述真正执行数据库操作通常发生在await session.execute(...) await session.scalars(...) await session.commit()二、安装异步ORM所需依赖本项目使用FastAPI、SQLAlchemy、MySQL和asyncmypython -m pip install fastapi uvicorn[standard] sqlalchemy asyncmy建议使用python -m pip 而不是直接使用pip这样可以尽量保证安装包的python与当前运行程序的python是同一个。检查解释器python -c import sys; print(sys.executable)检查驱动是否安装python -m pip show asyncmy三、创建异步数据库引擎数据库连接地址格式如下mysqlasyncmy://用户名:密码数据库地址:3306/数据库名?charsetutf8mb4为了避免密码泄露不建议把真实的连接地址写进代码可以从环境变量中读取import os from sqlalchemy.ext.asyncio import create_async_engine DATABASE_URL os.getenv(DATABASE_URL) if not DATABASE_URL: raise RuntimeError(请先配置 DATABASE_URL 环境变量) engine create_async_engine( DATABASE_URL, pool_size10, max_overflow20, pool_pre_pingTrue, echoTrue, )需要注意PyCharm 右侧的 Database 面板与代码中的 SQLAlchemy 引擎是两套独立配置四、设计ORM模型先导入需要的类型from datetime import datetime from decimal import Decimal from sqlalchemy import DateTime, Numeric, String, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column建立所有模型的基类class Base(DeclarativeBase): pass把创建时间提取成公共类class TimestampMixin: create_time: Mapped[datetime] mapped_column( DateTime, server_defaultfunc.now(), comment创建时间, ) update_time: Mapped[datetime] mapped_column( DateTime, server_defaultfunc.now(), onupdatefunc.now(), comment更新时间, )然后定义图书表class Book(TimestampMixin, Base): __tablename__ book id: Mapped[int] mapped_column( primary_keyTrue, autoincrementTrue, comment书籍 ID, ) bookname: Mapped[str] mapped_column( String(255), comment书名, ) author: Mapped[str] mapped_column( String(255), comment作者, ) price: Mapped[Decimal] mapped_column( Numeric(10, 2), comment价格, ) #两组类型 Mapped[str] # Python 中的字符串类型 String(255) # 数据库中的 VARCHAR(255) Mapped[datetime] # Python 中的日期时间类型 DateTime # 数据库中的日期时间类型func.now() 表示调用数据库的当前时间函数必须带括号价格使用ldecimal无精度问题。五、让FastAPI启动时自动建表首先定义建表函数async def create_tables(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)这里的Base.metadata是收集了所有继承Base的模型信息新版Fastapi推荐使用lifespan管理启动和关闭逻辑。from contextlib import asynccontextmanager from fastapi import FastAPI asynccontextmanager async def lifespan(app: FastAPI): # 应用启动时执行 await create_tables() yield # 应用关闭时执行 await engine.dispose() app FastAPI(lifespanlifespan)启动fastapi-调用create_tables()-检查并创建数据表-yield后开始接受请求-关闭时释放连接池六 、问什么退出代码为0表却没有创建直接文件只有类和定义函数python就不会启动fastapi可以在底部加上if __name__ __main__: import uvicorn uvicorn.run( app, host127.0.0.1, port8000, )或者将文件命名为main.py在终端直接运行python -m uvicorn orm_app:app --reload只有uvicorn真正启动Fastapi的生命周期函数才会执行。Application startup completeUvicorn running on http://127.0.0.1:8000七、创建异步sessionengine负责管理数据库连接session负责一次具体的数据库操作。from collections.abc import AsyncGenerator from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker SessionLocal async_sessionmaker( engine, expire_on_commitFalse, )定义fastapi依赖async def get_session() - AsyncGenerator[AsyncSession, None]: async with SessionLocal() as session: yield session然后再路由中注入from fastapi import Depends app.get(/example) async def example( session: AsyncSession Depends(get_session), ): return {message: 已经获得数据库 Session}八、定义接口请求模型ORM模型负责描述数据库表,pydantic模型负责校验接口数据两者职责不同。from decimal import Decimal from pydantic import BaseModel, ConfigDict, Field class BookCreate(BaseModel): bookname: str Field(min_length1, max_length255) author: str Field(min_length1, max_length255) price: Decimal Field(gt0, decimal_places2) class BookUpdate(BaseModel): bookname: str | None Field( defaultNone, min_length1, max_length255, ) author: str | None Field( defaultNone, min_length1, max_length255, ) price: Decimal | None Field( defaultNone, gt0, decimal_places2, ) class BookOut(BaseModel): model_config ConfigDict(from_attributesTrue) id: int bookname: str author: str price: Decimalfrom_attributesTrue允许pydantic从ORM对象的属性中读取数据。九、在路由中查询数据1.查询全部数据from sqlalchemy import select app.get(/books, response_modellist[BookOut]) async def get_books( session: AsyncSession Depends(get_session), ): stmt select(Book).order_by(Book.id.desc()) result await session.scalars(stmt) return result.all()2.根据主键查询from fastapi import HTTPException app.get(/books/{book_id}, response_modelBookOut) async def get_book( book_id: int, session: AsyncSession Depends(get_session), ): book await session.get(Book, book_id) if book is None: raise HTTPException( status_code404, detail书籍不存在, ) return book3.条件和比较查询# 查询价格大于 50 的图书 stmt select(Book).where(Book.price 50) # 查询价格在 50 到 100 之间的图书 stmt select(Book).where( Book.price 50, Book.price 100, ) # 查询指定 ID stmt select(Book).where( Book.id.in_([1, 2, 3]) ) # 查询作者不是“匿名”的图书 stmt select(Book).where( Book.author ! 匿名 )4.模糊查询stmt select(Book).where( Book.bookname.contains(Python) )常见模糊查询Book.bookname.contains(Python) Book.bookname.startswith(FastAPI) Book.bookname.endswith(实战) Book.bookname.like(%Python%)5.聚合查询统计图书数量、平均价格、最高和最低价格from sqlalchemy import func, select app.get(/books/statistics) async def book_statistics( session: AsyncSession Depends(get_session), ): stmt select( func.count(Book.id), func.avg(Book.price), func.max(Book.price), func.min(Book.price), ) result await session.execute(stmt) count, average, maximum, minimum result.one() return { count: count, average_price: float(average or 0), maximum_price: float(maximum or 0), minimum_price: float(minimum or 0), }十、分页查询分页的核心公式offset page-1*page_sizefrom fastapi import Query app.get(/books/page, response_modellist[BookOut]) async def get_books_by_page( page: int Query(1, ge1), page_size: int Query(10, ge1, le100), session: AsyncSession Depends(get_session), ): offset (page - 1) * page_size stmt ( select(Book) .order_by(Book.id.desc()) .offset(offset) .limit(page_size) ) result await session.scalars(stmt) return result.all()十一、新增数据app.post( /books, response_modelBookOut, status_code201, ) async def create_book( payload: BookCreate, session: AsyncSession Depends(get_session), ): book Book(**payload.model_dump()) session.add(book) await session.commit() await session.refresh(book) return book十二、更新数据app.put(/books/{book_id}, response_modelBookOut) async def update_book( book_id: int, payload: BookUpdate, session: AsyncSession Depends(get_session), ): book await session.get(Book, book_id) if book is None: raise HTTPException( status_code404, detail书籍不存在, ) update_data payload.model_dump( exclude_unsetTrue ) for field, value in update_data.items(): setattr(book, field, value) await session.commit() await session.refresh(book) return book十三、删除数据app.delete(/books/{book_id}) async def delete_book( book_id: int, session: AsyncSession Depends(get_session), ): book await session.get(Book, book_id) if book is None: raise HTTPException( status_code404, detail书籍不存在, ) await session.delete(book) await session.commit() return {message: 删除成功}新增、删除和修改都需要提交事务await session.commit()如果执行过程中发生异常应回滚await session.rollback()十四、我实际遇到的问题与排错思路1.asyncmy找不到先确认运行解释器python -c import sys; print(sys.executable)在使用同一个安装解释器安装from sqlalchemy import DateTime, String, func2.安装时发生SSL错误SSLError: EOF occurred in violation of protocol这说明pip没有正常连接 PyPI。后面的“找不到匹配版本”可能只是网络失败产生的连带信息。应优先检查当前网络和代理、系统时间、python证书、pycharm与终端解释器是否一致3.DateTime、String、func报红需要从sqlalchemy导入from sqlalchemy import DateTime, String, func4.func.now报错需要调用函数func.now() 不能只写func.now5.表没有创建检查数据库端口是否可以访问-用户名、密码、数据库名是否正确-Book是否继承Base-是否使用Base.metadata.create_all -metadata是否拼写正确-lifespan是否绑定到FastAPI-是否启动7.修改模型后数据库字段没有变化Base.metadata.create_all只创建不存在的表不负责修改现有的表结构。学习阶段可以删除测试表后重新创建正式项目应该使用Alembic管理数据库迁移。十五、总结第一次解除orm感觉复杂并不是因为book表本事复杂而是这三张表同时存在这三个系统中MySQL 数据库、SQLAlchemy 异步 ORM、FastAPI Web 应用可以把代码分成三层配置层Engine、SessionLocal、lifespan模型层Base、Book、Pydantic Schema接口层查询、新增、更新、删除配置层通常只需要写一次。后续开发主要是在模型层添加字段在接口层编写业务逻辑。这次学习让我形成了一套比较实用的排错逻辑先检查解释器和依赖-再检查数据库网络和账号-在检查模型和metadata-在检查lifespan是否绑定-最后确认uvicorn是否真正启动按照这样的链路排查比看到最后一行错误就不断的修改模型更有效。ORM也不是让开发者完全忘记SQL而是把表结构、事务和业务操作组成成更容易维护的python代码。当engine、model、session和fastapi生命周期真正连接起来以后后续的条件查询、分页和增删改本质上都是在同一条链路上继续向前。