【 FastAPI 】ORM

**FastAPI ORM ** 一、ORM 是什么对象关系映射Object-Relational Mapping, ORM是一种编程技术用于在面向对象语言如 Python与关系型数据库如 MySQL之间建立映射桥梁。“它允许开发者通过操作对象的方式与数据库进行交互并且无需编写复杂的 SQL 语句”举例不用写SELECT * FROM book WHERE name Python可以直接写db.query(Book).filter(Book.name Python).first()二、ORM 使用五大步骤步骤内容1. 导入模块标准库导入2. 创建引擎异步 池配置3. 定义模型类结构清晰继承统一4. 启动建表自动建表无需手动5. 操作数据增删改查全功能实现--- ## 步骤 1导入 ORM 模块 python from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy import DateTime, func, String, Float, select from sqlalchemy.sql.functions import user说明create_async_engine创建异步数据库引擎async_sessionmaker会话工厂AsyncSession异步会话对象DeclarativeBaseORM 基类Mapped/mapped_column定义模型字段select()SQL 查询构造器func.now()数据库函数自动获取当前时间步骤 2创建异步数据库引擎# 数据库连接 URL 格式 驱动: //用户名 密码 主机 端口 /数据库名?字符集ASYNC_DATABASE_URLmysqlaiomysql://root:123456localhost:3306/FastAPI_first?charsetutf8mb4# 创建异步数据库引擎管理数据库连接池async_enginecreate_async_engine(ASYNC_DATABASE_URL,echoTrue,# 调试用打印所有 SQL 语句pool_size20,# 连接池大小max_overflow10,# 超出后最多额外创建 10 个连接pool_recycle3600# 连接闲置超 1 小时自动回收)关键点mysqlaiomysql使用异步驱动charsetutf8mb4支持 emoji 与多语言echoTrue仅限开发环境生产建议设为False警告pool_size20过大生产环境建议设为10步骤 3定义模型类基类Base统一时间字段classBase(DeclarativeBase):ORM 基类所有模型类都继承自此类create_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now(),comment创建时间)update_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultfunc.now(),onupdatefunc.now(),comment更新时间)书籍模型类 → 对应book表classBook(Base):书籍模型对应数据库中的 book 表__tablename__book# mapped_column() 定义数据库列的属性id:Mapped[int]mapped_column(primary_keyTrue,commentID)name:Mapped[str]mapped_column(String(255),comment书名)author:Mapped[str]mapped_column(String(255),comment作者)price:Mapped[float]mapped_column(Float,comment价格)publisher:Mapped[str]mapped_column(String(255),comment出版社)用户模型类 → 对应user表classUser(Base):用户模型对应数据库中的 user 表__tablename__userid:Mapped[int]mapped_column(primary_keyTrue,commentID)username:Mapped[str]mapped_column(String(255),comment用户名)password:Mapped[str]mapped_column(String(255),comment密码)email:Mapped[str]mapped_column(String(255),comment邮箱)# 继承 Base 的 create_time / update_time所有模型继承 Base自动获得 create_time 和 update_time。步骤 4启动时创建表目标应用启动时自动根据模型类生成数据库表结构。asyncdefcreate_table():在应用启动时自动创建数据库表asyncwithasync_engine.begin()asconn:awaitconn.run_sync(Base.metadata.create_all)# run_sync: 运行同步方法# Base.metadata.create_all: 根据模型类自动创建所有表# 注册启动事件应用启动时自动执行 create_table()app.on_event(startup)asyncdefstartup_event():FastAPI 应用启动时触发的事件print(正在初始化数据库...)awaitcreate_table()print(数据库初始化完成)startup事件应用启动自动建表使用了async_engine.begin()run_sync()实现异步建表最佳实践避免手动运行init_db.py步骤 5操作数据TODO: 在这里添加 CRUD 操作的路由1、创建会话工厂2、编写依赖项获取数据库会话3、实现增删改查接口1. 创建会话工厂Async_sessionLocalasync_sessionmaker(bindasync_engine,#绑定数据库引擎class_AsyncSession,#指定会话类expire_on_commitFalse# 关键必须设置 #提交事务后session 指定会话不过期 不会重新查询数据库)**expire_on_commitFalse是核心灵魂**如果不设提交事务后对象失效 → 后续无法访问属性2. 编写依赖项get_db()asyncdefget_db():获取数据库会话asyncwithAsync_sessionLocal()assession:try:yieldsessionexceptException:awaitsession.rollback()raisefinally:awaitsession.close()使用了yieldasync with实现自动创建会话自动关闭异常时自动回滚通过Depends传递给路由函数3. 实现 CRUD 路由路由功能使用方法/book/books获取所有书籍await db.execute(select(Book))/book/book获取第一本.limit(1).first()/book/book/{id}按 ID 查询where(Book.id id)/book/book_by_name按名称查询where(Book.name name)/book/search_book搜索价格 ≥ 200where(Book.price 200)/user/users获取所有用户select(User)所有操作使用await db.execute(select(...))异步安全app.get(/book/books)asyncdefget_book_list(db:AsyncSessionDepends(get_db)):#Depends(get_db) 告诉 FastAPI这个参数依赖于 get_db() 函数的返回值FastAPI 会自动调用 get_db()并将结果注入到 db 参数获取所有书籍列表# 查询操作 await db.excute(select(模型类)) 返回一个 ORM 对象# 获取所有数据 scalars(),all()# 获取指定数据# 1、scalars().first()# 2、get(模型类主键值)resultawaitdb.execute(select(Book))# 等待这个异步操作执行完毕然后获取结果bookresult.scalars().all()# .scalars() - 从 Row 对象中提取出 Book 模型对象.all() - 将所有 Book 对象收集到一个列表中# 将 SQLAlchemy 对象转换为字典列表returnbook# 获取第一个书籍app.get(/book/book)asyncdefget_book(db:AsyncSessionDepends(get_db)):获取第一个书籍resultawaitdb.execute(select(Book).limit(1))bookresult.scalars().first()returnbookapp.get(/book/book/{id})asyncdefget_book_by_id(id:int,db:AsyncSessionDepends(get_db)):获取指定 ID 的书籍resultawaitdb.execute(select(Book).where(Book.idid))bookresult.scalars().first()returnbookapp.get(/book/book_by_name)asyncdefget_book_by_name(name:str,db:AsyncSessionDepends(get_db)):获取指定名称的书籍resultawaitdb.execute(select(Book).where(Book.namename))bookresult.scalars().first()returnbookapp.get(/book/search_book)asyncdefsearch_book(name:str,db:AsyncSessionDepends(get_db)):搜索书籍resultawaitdb.execute(select(Book).where(Book.price200))bookresult.scalars().all()returnbookapp.get(/user/users)asyncdefget_user_list(db:AsyncSessionDepends(get_db)):获取所有用户列表resultawaitdb.execute(select(User))userresult.scalars().all()returnuser三、重要的知识点你写的补充说明scalars().all()从Row对象中提取出Book实例不是Row.scalars().first()获取第一个对象返回Book实例where(...)条件查询避免 SQL 注入func.now()数据库时间函数自动获取当前时间expire_on_commitFalse核心必须设置否则对象失效警告如果expire_on_commitTrue你执行了await db.commit()后再尝试.name会报错AttributeError: Book object has no attribute name