Python（FastAPI） 路由匹配中使用ORM实现查询功能

从建表到路由到查询代码fromfastapiimportFastAPI,DependsfromdatetimeimportdatetimefromsqlalchemyimportString,DateTime,func,selectfromsqlalchemy.ext.asyncioimportcreate_async_engine,async_sessionmaker,AsyncSessionfromsqlalchemy.ormimportDeclarativeBase,Mapped,mapped_column appFastAPI()# 1.创建异步引擎#ASYNC_DATAASE_URL为数据库的地址ASYNC_DATAASE_URLmysqlaiomysql://root:630229localhost:3306/fast_api_db?charsetutf8async_enginecreate_async_engine(ASYNC_DATAASE_URL,echoTrue,# 可选删除SQL日志pool_size10,#设置连接池活跃的连接数max_overflow20,# 允许额外的连接数)# 2.定义模型类: 基类表对应的模型类# 基类:创建时间、更新时间;书籍表:id、书名、作者、价格、出版社classBase(DeclarativeBase):create_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),defaultdatetime.now,comment创建时间)update_time:Mapped[datetime]mapped_column(DateTime,insert_defaultfunc.now(),onupdatefunc.now(),defaultdatetime.now,comment修改时间)classBook(Base):__tablename__bookid:Mapped[int]mapped_column(primary_keyTrue,comment书籍id)bookname:Mapped[str]mapped_column(String(255),comment书名)author:Mapped[str]mapped_column(String(255),comment作者)price:Mapped[float]mapped_column(comment价格)publisher:Mapped[str]mapped_column(String(255),comment出版社)# 3.启动应用时建表FASTAPI启动时调用建表的函数asyncdefcreate_tables():#创建异步引擎创建事务建表asyncwithasync_engine.begin()asconn:awaitconn.run_sync(Base.metadata.create_all)#模型类的元数据app.on_event(startup)asyncdefstartup_event():awaitcreate_tables()app.get(/)asyncdefroot():return{message:Hello World}# 路由匹配中使用 ORM ,创建依赖项使用 Depends注入到处理函数# 需求查询功能的接口查询图书》依赖注入创建依赖项获取数据库会话Depends注入路由处理函数# 4.创建异步会话工厂AsyncSessionLocalasync_sessionmaker(bindasync_engine,# 绑定异步引擎class_AsyncSession,# 设置会话类为异步会话expire_on_commitFalse# 设置会话提交时是否自动刷新)# 5.依赖项获取数据库会话asyncdefget_database():asyncwithAsyncSessionLocal()assession:try:yieldsession# 返回数据库会话awaitsession.commit()# 提交事务exceptException:awaitsession.rollback()# 回滚事务raisefinally:awaitsession.close()# 关闭会话# 6.路由查询图书app.get(/book/books)asyncdefget_books_list(session:AsyncSessionDepends(get_database)):resultawaitsession.execute(select(Book))booksresult.scalars().all()returnbooks完美