python之FastAPI进阶篇

python之fastAPI进阶篇

1.中间件

作用：为每个接口前后添加统一的处理逻辑，例如多个接口都要进行身份验证，记录日志等

多个中间件自下而上进行执行

@app.middleware("http")
async def middleware(request,call_next):
    response=await call_next(request)
return response

2.依赖注入

为什么要依赖注入？减少代码重复

什么是依赖注入？可重复使用的组件

比如写多个分页列表，参数都是固定，所以每次都得写一遍

@app.get("/newslist")
def newslist(skip:int=Query(0,ge=0),limit:int=Query(10,le=60)):
return {"list":"新闻列表"}

使用依赖项

1.创建依赖项

async def common_parameter(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("/newslist")
def newslist(common=Depends(common_parameter)):
return common

4.应用场景

1.处理请求参数

2.共享业务逻辑

3.共享数据库连接

4.安全与认证

ORM

用于在面向对象编程语言和关系型数据库之间建立映射。它允许开发者通过操作对象的方式与数据库进行交互，而无需直接编写复杂的SQL语句。

如何使用？

1.创建sql引擎

pip install sqlalchemy

from sqlalchemy.ext.asyncio import create_async_engine

ASYNC_DATABASE_URL="mysql+aiomysql://root: @localhost:3306/test?charset=utf8"

async_engine=create_async_engine(
    ASYNC_DATABASE_URL,
echo=True,
    pool_size=10,
    max_overflow=20
)

2.定义模型类

#基类
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(), onupdate=func.now(), default=datetime.now, comment="修改时间")
#数据表类
class User(Base):
    __tablename__ = "user"
    id:Mapped[int]=mapped_column(primary_key=True)
    name:Mapped[str]=mapped_column(String(255))

3.启动时建表

async def create_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@asynccontextmanager
async  def lifespan(app:FastAPI):
print("启动前执行")
    await create_tables()
    yield
print("关闭后前执行")
app=FastAPI(lifespan=lifespan)

路由匹配中使用ORM

创建依赖项，使用Depends注入到处理函数

#异步会话工厂
AsyncSessionLocal=async_sessionmaker(
bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)
#依赖项：获取数据库会话
async  def getdatabase():
    async  with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

在路由中使用

@app.get("/userlist")
async def userlist(db:AsyncSession=Depends(getdatabase)):
    result=await db.execute(select(User))
    userlist=result.scalars().all()
return  userlist

获取单条数据

@app.get("/userone")
async def userone(db:AsyncSession=Depends(getdatabase)):
    result=await db.get(User,1)
return result

数据库查询

@app.get("/user/{id}")
async def userone(id:int,db:AsyncSession=Depends(getdatabase)):
    result=await db.execute(select(User).where(User.id==id))
return result.scalar_one_or_none()

模糊查询

@app.get("/book/get_books")
async def get_book_list(db: AsyncSession = Depends(getdatabase)):
    result = await db.execute(select(User).where(User.name.like("朱%")))
    book = result.scalars().all()
return book

精准查询

@app.get("/users/getuser")
async def getuser(db: AsyncSession = Depends(getdatabase)):
    result = await db.execute(select(User).where(User.name=='朱广鑫'))
    book = result.scalars().all()
return book

包含查询

@app.get("/users/getuserById")
async def getuserById(db: AsyncSession = Depends(getdatabase)):
    list=[1,2]
    result = await db.execute(select(User).where(User.id.in_(list)))
    book = result.scalars().all()
return book

聚合查询

count：统计行数量

avg：求平均值

max：求最大值

min：求最小值

sum：求和

@app.get("/food/getfoodcout")
async def getfoodcout(db: AsyncSession = Depends(getdatabase)):
    result = await db.execute(select(func.sum(Food.price)))
    count = result.scalar()
return count

分页查询

@app.get("/food/getpage")
async def getpage(page: int = 1,page_size: int = 3,db: AsyncSession = Depends(getdatabase)):
    skip = (page - 1) * page_size
    stmt = select(Food).offset(skip).limit(page_size)
    result = await db.execute(stmt)
    food = result.scalars().all()
return {"books": food}

新增

@app.post("/food/addfood")
async def addfood(food:FoodBase,db: AsyncSession = Depends(getdatabase)):
    food_obj=Food(**food.__dict__)
    db.add(food_obj)
    await db.commit()
return food

修改

@app.post("/food/updatefood/{id}")
async def updatefood(id:int,data:FoodBaseUpdate,db: AsyncSession = Depends(getdatabase)):
    food = await db.get(Food, id)
if food is None:
        raise HTTPException(status_code=404, detail="Book not found")
# 2. 修改属性（重新赋值）
    food.name = data.name
    food.price=data.price
    food.remark=data.remark
# 3. 提交
    await db.commit()
return food

删除

@app.delete("/food/delete_food/{id}")
async def delete_food(id: int, db: AsyncSession = Depends(getdatabase)):
    food = await db.get(Food, id)
if food is None:
        raise HTTPException(status_code=404, detail="Book not found")
    await db.delete(food)
    await db.commit()
return {"message": "food deleted"}