中间件

中间件是一个再每次请求进入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,ge=0), limit: int = Query(10,le=60) ): return { "skip":skip, "limit":limit }

2.导入Depends

from fastapi import Depends

3.声名依赖项

@app.get("news/news_list") async def get_news_list( commons = Depends(common_parameters) ): return commons

ORM简介

ORM是一种操作数据库的编程技术，用于再面向对象编程语言和关系数据库之间建立映射

使用流程

使用ORM

创建异步数据库引擎

from sqlalchemy.ext.asyncio import create_async_engine # 异步MySQL连接地址（驱动为aiomysql） ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/lol?charset=utf8" # 创建异步数据库引擎 async_engine = create_async_engine( ASYNC_DATABASE_URL, echo=True, # 可选：打印执行的原生SQL日志，开发调试开启，生产关闭 pool_size=10, # 连接池常驻持久连接数量 max_overflow=20 # 连接池额外临时扩容最大连接数 )

定义模型类

包含

1.基类 -- 包含通用属性字段的映射

class Base(DeclarativeBase): create_time: Mapped[datetime] = mapped_column( Datetime, insert_default=func.now(),default=datetime.now,comment="创建时间") update_time: Mapped[datetime] = mapped_column( Datetime, insert_default=func.now(),default=datetime.now,comment="修改时间")

2.定义数据表对应的模型类

class Book(Base): __tablename__="book" id:Mapped[int] = mapped_column(primary_key=True) 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( bind=async_engine,#绑定数据引擎 class_=AsyncSession,#制定会话类 expire_on_commit=False#会话对象不过期，不重新查询数据库 ) 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_model=list[BookSchema]) async def get_book_list( author: str = Query(None, description="Filter by author"), bookname: str = Query(None, description="Filter 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()