一、为什么要学ORM半个月前刚开始学FastAPI的时候我练习时写的接口都是返回假数据比如app.get(/users)asyncdefget_users():return[{id:1,name:张三}]实际项目肯定是需要从数据库读数据的。我开始啥也不会就直接写SQL字符串比如sqlfSELECT * FROM t_user WHERE id {user_id}后来进一步学习相关知识时了解到这样做会有SQL注入风险而且每个接口都要重复写连接、关闭的逻辑代码很冗余也不想敲。然后就知道了ORM这个概念。ORM的意思很简单就是把数据库的表当成Python的类把表里的每一行数据当成这个类的一个对象。这样我在操作Python对象时ORM自动帮我翻译成SQL语句。二、安装依赖我学习时用的MySQL数据库做这个的时候需要安装两个包pipinstallsqlalchemy[asyncio]aiomysqlsqlalchemy[asyncio]SQLAlchemy的异步版本aiomysql异步MySQL驱动这里有个小坑[asyncio]不是文件名的一部分而是告诉pip额外安装异步相关的依赖。我第一次搞的时候没加这个方括号结果出问题代码跑不起来。三、连接数据库3.1 创建异步引擎引擎就是连接数据库的入口。异步场景下要用create_async_enginefromsqlalchemy.ext.asyncioimportcreate_async_engine DATABASE_URLmysqlaiomysql://root:rootlocalhost:3306/fastapi_review?charsetutf8enginecreate_async_engine(DATABASE_URL,echoTrue,# 打印SQL语句学习阶段一定要开pool_size10,# 连接池大小max_overflow20# 最大额外连接数)echoTrue这个参数特别有用它会在控制台打印所有生成的SQL。我学习的时候就喜欢用它这样就能直观的看到ORM到底翻译成了什么样的SQL。3.2 应用启动时建表我需要在服务启动的时候自动在数据库里创建表。代码如下app.on_event(startup)asyncdefinit():awaitcreate_tables()asyncdefcreate_tables():asyncwithengine.begin()asconn:awaitconn.run_sync(Base.metadata.create_all)这里我一开始不明白为什么要用run_sync后来查资料才明白Base.metadata.create_all是SQLAlchemy早期的同步API。我现在用的是异步引擎所以需要用run_sync把它包一层让它能在异步环境里执行。简单说就是——异步引擎调用同步方法需要这个桥接。四、定义模型4.1 基类设计所有模型都要继承一个基类我把它设计成了这样fromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_columnfromsqlalchemyimportDateTime,funcfromdatetimeimportdatetimeclassBase(DeclarativeBase):passDeclarativeBase是SQLAlchemy 2.0的新写法取代了旧的declarative_base()函数。我觉得这种写法更直观。我还给基类加了两个自动维护的时间字段classBase(DeclarativeBase):# 创建时间插入时自动填充create_time:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now)# 更新时间插入和更新时自动刷新update_time:Mapped[datetime]mapped_column(DateTime,defaultdatetime.now,onupdatefunc.now(),insert_defaultfunc.now())func.now()是SQL函数让数据库自己算当前时间比Python的datetime.now更准确。这样我就不用每次插入数据时手动填时间了。4.2 用户模型我的用户模型长这样classUser(Base):__tablename__t_userid:Mapped[int]mapped_column(primary_keyTrue,autoincrementTrue,nameuser_id,comment用户ID)name:Mapped[str]mapped_column(String(20),nullableFalse,nameuser_name,comment用户名称)password:Mapped[str]mapped_column(String(20),nullableFalse,nameuser_password,comment用户密码)salary:Mapped[float]mapped_column(Float(6,2),nullableFalse,nameuser_salary,comment用户薪水)birthday:Mapped[datetime]mapped_column(DateTime,nullableFalse,nameuser_birthday,comment用户出生日期)这里有几个我踩过的坑__tablename__必须写不然SQLAlchemy不知道表名叫什么nameuser_id是指定数据库里的真实列名。如果不写默认就用Python属性名idMapped[int]这种写法是SQLAlchemy 2.0的类型注解风格比旧版更清晰Float(6, 2)表示总共6位数字小数占2位比如1234.56五、会话管理我最花时间的部分5.1 创建会话工厂fromsqlalchemy.ext.asyncioimportasync_sessionmaker,AsyncSession AsyncSessionLocalasync_sessionmaker(bindengine,class_AsyncSession,expire_on_commitFalse)expire_on_commitFalse这个参数很重要。如果不加提交后再访问对象属性可能会报错。异步场景下建议一定要设成False。5.2 依赖注入函数这部分我研究了很久最终写成这样asyncdefget_session():asyncwithAsyncSessionLocal()assession:try:yieldsessionawaitsession.commit()exceptException:awaitsession.rollback()finally:awaitsession.close()为什么用yield这是FastAPI依赖注入的固定写法yield之前的代码请求进来时执行创建会话yield session把会话交给路由函数使用yield之后的代码请求结束后执行提交或回滚然后关闭try-except-finally确保不管发生什么连接最终都会关闭。我一开始没加finally后来意识到如果代码中途报错连接就泄漏了。使用的时候很简单app.get(/users)asyncdefget_user_list(session:AsyncSessionDepends(get_session)):...FastAPI会自动调用get_session()把生成的session传进来。六、查询操作我写了5种场景6.1 查询全部app.get(/users)asyncdefget_user_list(session:AsyncSessionDepends(get_session)):stmtSelect(User)resultawaitsession.execute(stmt)user_listresult.scalars().all()returnuser_listscalars().all()返回所有记录的对象列表。如果只想取一条用scalar()。6.2 精确查询app.get(/users/)asyncdefget_user_by_name(name:str,session:AsyncSessionDepends(get_session)):stmtSelect(User).where(User.namename)resultawaitsession.execute(stmt)returnresult.scalars().all()6.3 模糊查询app.get(/users/like/)asyncdefget_user_like_name(name:str,session:AsyncSessionDepends(get_session)):stmtSelect(User).where(User.name.like(f%{name}%))resultawaitsession.execute(stmt)returnresult.scalars().all()like是SQL的模糊匹配%是通配符。6.4 批量查询INapp.get(/users/in/)asyncdefget_user_in_ids(ids:str,session:AsyncSessionDepends(get_session)):id_list[int(item)foriteminids.split(,)]stmtSelect(User).where(User.id.in_(id_list))resultawaitsession.execute(stmt)returnresult.scalars().all()注意SQLAlchemy里用in_()后面有个下划线因为in是Python关键字。6.5 范围查询BETWEENapp.get(/users/between/{min_salary}/{max_salary})asyncdefget_user_between_salary(min_salary:float,max_salary:float,session:AsyncSessionDepends(get_session)):stmtSelect(User).where(User.salary.between(min_salary,max_salary))resultawaitsession.execute(stmt)returnresult.scalars().all()6.6 主键快捷查询app.get(/users/{id})asyncdefget_user_by_id(id:int,session:AsyncSessionDepends(get_session)):userawaitsession.get(User,id)returnusersession.get(User, id)是最简洁的主键查询方式等同于SELECT * FROM t_user WHERE user_id ?。6.7 分页查询app.get(/users/page/{page_no}/{page_size})asyncdefget_user_by_page(page_no:int,page_size:int,session:AsyncSessionDepends(get_session)):stmtSelect(User).offset((page_no-1)*page_size).limit(page_size)resultawaitsession.execute(stmt)returnresult.scalars().all()分页公式跳过(page_no - 1) * page_size条取page_size条。七、增删改7.1 添加数据我先用Pydantic定义了请求模型classUserRequest(BaseModel):name:strpassword:strsalary:floatbirthday:datetime然后这样添加app.post(/users)asyncdefadd_user(user:UserRequest,session:AsyncSessionDepends(get_session)):user_objectUser(nameuser.name,passworduser.password,salaryuser.salary,birthdayuser.birthday)session.add(user_object)return{code:200,message:添加成功}注意我没有手动commit()因为get_session在yield之后会自动提交。7.2 删除数据app.delete(/users/{id})asyncdefdelete_user(id:int,session:AsyncSessionDepends(get_session)):stmtDelete(User).where(User.idid)awaitsession.execute(stmt)awaitsession.commit()return{code:200,message:删除成功}这里我手动commit()了因为删除操作我想确保立即生效。顺便提一下我在注释里写了实际项目中删除基本都是逻辑删除改个状态字段比如is_deleted True不会真的删掉数据。这个我记着了以后做项目要注意。7.3 更新数据app.put(/users/{id})asyncdefupdate_user(id:int,user_request:UserRequest,session:AsyncSessionDepends(get_session)):stmtSelect(User).where(User.idid)resultawaitsession.execute(stmt)userresult.scalar_one_or_none()user.nameuser_request.name user.passworduser_request.password user.salaryuser_request.salary user.birthdayuser_request.birthdayreturn{code:200,message:修改成功}SQLAlchemy的更新很Pythonic直接改对象属性提交时ORM会自动生成UPDATE语句。八、我踩过的坑坑现象解决忘记await报错coroutine object所有session.execute()都要加awaitin写成in语法错误SQLAlchemy里是in_()带下划线会话没关闭连接池耗尽用async with或yieldfinallycommit时机混乱数据没写入统一交给get_session管理特殊情况再手动commit更新时没查对象不知道改谁先Select查出对象再改属性九、总结学完这些知识后我个人感受是ORM确实省事不用自己手写SQL操作Python对象就行异步要处处加await从引擎到会话到执行全程异步会话生命周期最重要打开→使用→提交/回滚→关闭这个流程不能乱PydanticORM是绝配前端传JSON → Pydantic校验 → 转ORM对象 → 入库下一步我打算学把代码拆到不同文件config/models/schemas/dependencies/routers表与表之间的关联比如用户和订单的一对多关系Alembic数据库迁移工具