中间件
中间件是一个再每次请求进入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 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 = "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()
