FastApi进阶

中间件中间件是一个再每次请求进入fastapi时都会执行的函数他在请求到达实际路径操作之前执行并且再相应返回客户端之前再运行一次执行顺序按代码顺序自底向上执行具体使用app.middleare(http) async def middleare(request,call_next): print(中间件开始处理 -- start) response await call_next(request) print(中间件处理完成 -- end) return response依赖注入系统可以用来共用通用逻辑减少代码重复具体场景如下使用步骤1.创建依赖项async def common_paremeters( skip: int Query(0,ge0), limit: int Query(10,le60) ): return { skip:skip, limit:limit }2.导入Dependsfrom fastapi import Depends3.声名依赖项app.get(news/news_list) async def get_news_list( commons Depends(common_parameters) ): return commonsORM简介ORM是一种操作数据库的编程技术用于再面向对象编程语言和关系数据库之间建立映射使用流程使用ORM创建异步数据库引擎from sqlalchemy.ext.asyncio import create_async_engine # 异步MySQL连接地址驱动为aiomysql ASYNC_DATABASE_URL mysqlaiomysql://root:123456localhost:3306/lol?charsetutf8 # 创建异步数据库引擎 async_engine create_async_engine( ASYNC_DATABASE_URL, echoTrue, # 可选打印执行的原生SQL日志开发调试开启生产关闭 pool_size10, # 连接池常驻持久连接数量 max_overflow20 # 连接池额外临时扩容最大连接数 )定义模型类包含1.基类 -- 包含通用属性字段的映射class Base(DeclarativeBase): create_time: Mapped[datetime] mapped_column( Datetime, insert_defaultfunc.now(),defaultdatetime.now,comment创建时间) update_time: Mapped[datetime] mapped_column( Datetime, insert_defaultfunc.now(),defaultdatetime.now,comment修改时间)2.定义数据表对应的模型类class Book(Base): __tablename__book id:Mapped[int] mapped_column(primary_keyTrue) bookname:Mapped[str] mapped_column(String(255)) author:Mapped[str] mapped_column(String(255))3.创建数据库表async def create_tables(): async with async_engine.begin() as conn: await conn.run_sync(Base.matadata.create_all) app.on_event(startup) async def startup_event(): await create_tables()路由匹配中使用ORM核心创建依赖项使用Depends注入处理函数AsyncSessionLocal async_sessionmaker( bindasync_engine,#绑定数据引擎 class_AsyncSession,#制定会话类 expire_on_commitFalse#会话对象不过期不重新查询数据库 ) async def get_db(): async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception as e: await session.rollback() raise e finally: await session.close()使用依赖app.get(/book/news_list) async def get_book(db: AsyncSession Depends(get_db)): result await db.execute(select(Book)) return result.scalars().all()数据库操作查询核心语句await db.execute(select(模型类))#返回一个ORM对象 #获取所有数据 scalars().all() #获取单条数据 scalars().first() get(模型类主键值)条件查询app.get(/book/list, response_modellist[BookSchema]) async def get_book_list( author: str Query(None, descriptionFilter by author), bookname: str Query(None, descriptionFilter by bookname), db: AsyncSession Depends(get_db) ): query select(Book) if author: query query.where(Book.author.like(f%{author}%)) if bookname: query query.where(Book.bookname.like(f%{bookname}%)) result await db.execute(query) return result.scalars().all()