中间件

中间件的作用：

• 预处理请求：例如检查认证头、解析 IP、记录日志、修改请求数据。

• 后处理响应：例如添加响应头（如 CORS）、压缩内容、记录响应时间。

• 全局拦截：统一处理异常、实现限流、黑白名单。

• 性能监控：统计每个请求的处理耗时。

中间件的实现：

使用 @app.middleware("http") 装饰器来实现

@app.middleware("http")
async def middleware_one(request, call_next):
    print("before 1")
    response = await call_next(request)  # await是异步处理
    print("after 1")
    return response

中间件的执行顺序：

按代码自下而上执行

例如：

@app.middleware("http")
async def middleware_one(request, call_next):
    print("before 1")
    response = await call_next(request)
    print("after 1")
    return response

@app.middleware("http")
async def middleware_two(request, call_next):
    print("before 2")
    response = await call_next(request)
    print("after 2")
    return response

访问时输出顺序：

before 1
before 2
(路由处理)
after 2
after 1

依赖注入

FastAPI中，依赖注入通过抽取可复用的组件，实现代码复用、解耦且可轻松替换依赖项进行测试

怎么使用依赖注入系统?

创建依赖    ->    项导入Depends    ->    声明依赖项

from fastapi import Depends
async def common_parameters(
        skip: int = Query(0, ge=0),
        limit: int = Query(10, le=60)
    ):
    return {"skip": skip, "limit": limit}

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

ORM

SQLAlchemy ORM安装：

在命令行中执行下面代码

pip install sqlalchemy[asyncio]

建表：

第一步：创建异步引擎

mysql+aiomysql://账户:密码@localhost:3306/数据库名称?charset=utf8

ASYNC_DATABASE_URL = "mysql+aiomysql://root:root@localhost:3306/fastapi_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选，输出SQL日志
    pool_size=10,  # 设置连接池活跃的连接数
    max_overflow=20  # 允许额外的连接数
)

注意： 如果没有安装aiomysql需要先安装

pip install aiomysql

第二步：定义模型类:基类+表对应的模型类

基类:创建时间、更新时间，继承 DeclarativeBase(包含通用属性和字段的映射)

书籍表:id、书名、作者、价格、出版社

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


class Book(Base):
    __tablename__ = "book"

    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="者")
    price: Mapped[float] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")

第三步：建表，定义函数建表 FastAPI 启动的时候调用建表的函数

async def create_tables():
    # 获取异步引擎，创建事务-建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Base 模型类的元数据创建

# 该注解表示在启动FastAPI时调用该接口
@app.on_event("startup")
async def startup_event():
    await create_tables()

async with async_engine.begin() as conn:

• async_engine 是一个异步 SQLAlchemy 引擎实例，通常通过 create_async_engine() 创建。

• begin() 是一个异步上下文管理器，它会：

  • 获取一个数据库连接（从连接池中获取）。

  • 开始一个事务。

  • 在 async with 块结束时，如果块内没有发生异常，则自动提交事务；如果发生异常，则自动回滚事务。

• 返回的 conn 是一个异步连接对象（AsyncConnection），可以用于执行异步数据库操作。

因此，async with async_engine.begin() as conn: 确保了我们拥有一个干净的、处于事务中的数据库连接，所有在块内执行的 SQL 操作都会被包含在同一个事务中，要么全部成功提交，要么全部失败回滚。

await conn.run_sync(Base.metadata.create_all)

• conn.run_sync() 是一个特殊的方法，它允许我们在异步连接中运行一个同步函数。因为 SQLAlchemy 的核心（如 metadata.create_all）原本是同步的，在异步环境中需要包装才能执行。

• run_sync 会将该同步函数放到一个线程池中执行（以避免阻塞事件循环），并等待其完成。

• Base.metadata.create_all 是 SQLAlchemy 提供的同步方法，它会：

  • 检查数据库中是否已存在 Base.metadata 中定义的所有表。

  • 如果表不存在，则根据模型定义生成 CREATE TABLE 语句并执行（通过当前的连接）。

  • 如果表已存在，则默认不做任何操作（不会修改表结构，也不会删除数据）。

• Base 是我们定义的 ORM 基类（继承自 DeclarativeBase），它的 metadata 属性收集了所有继承自它的模型类的表定义。

因此，这行代码的作用是：在当前的数据库连接中，以同步方式执行表的创建操作，确保所有模型对应的表都被创建。

运行代码即可成功建表

路由匹配中使用ORM：

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

第一步：创建异步会话工厂

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,  # 绑定数据库引擎
    class_=AsyncSession,  # 指定会话类
    expire_on_commit=False  # 提交后会话不过期，不会重新查询数据库
)

第二步：创建依赖项，用于获取数据库会话

async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # 返回数据库会话给路由处理函数
            await session.commit()  # 提交事务
        except Exception:
            await session.rollback()  # 有异常，回滚
            raise
        finally:
            await session.close()  # 关闭会话

第三步：Depends 注入路由处理函数

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
    # 查询
    result = await db.execute(select(Book))
    book = result.scalars().all()
    return book

ORM增删改查：

查：

查询所有数据

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
    # 查询
    result = await db.execute(select(Book))
    book = result.scalars().all()
    return book

查询第一条数据

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
    # 查询
    result = await db.execute(select(Book))
    book = result.scalars().first()
    return book

根据主键获取单条数据

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
    # 查询
    book = await db.get(Book,5)  # 获取id为5的数据
    return book

通过where进行条件查询

@app.get("/book/get_book/{book_id}")
async def get_book_list(book_id: int, db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.id == book_id))
    book = result.scalar_one_or_none()    # 若有就返回查询到的值，若没有则返回null
    return book

like()模糊查询

result = await db.execute(select(Book).where(Book.author.like("曹%"))) # 查询作者以曹开头的数据

&和|查询

result = await db.execute(select(Book).where((Book.author.like("曹%"))&(Book.price > 100)))  # 返回两个条件都成立的数据
result = await db.execute(select(Book).where((Book.author.like("曹%"))|(Book.price > 100)))  # 返回两个条件成立其中一个的数据

in查询：将id_list中存在的id查询出来(如果数据库中不存在该id则不返回)

id_list = [1, 3, 5, 7]
result = await db.execute(select(Book).where(Book.id.in_(id_list)))

聚合计算：func.方法(模型类.属性)

• count: 统计行数量

• avg: 求平均值

• max: 求最大值

• min: 求最小值

• sum: 求和

@app.get("/book/count")
async def get_count(db: AsyncSession = Depends(get_database)):
    # result = await db.execute(select(func.count(Book.id)))
    # result = await db.execute(select(func.max(Book.price)))
    # result = await db.execute(select(func.sum(Book.price)))
    # result = await db.execute(select(func.avg(Book.price)))
    count = result.scalar()
    return count

分页查询：

select().offset().limit()

• offset: 跳过的记录数

• limit: 返回的记录数

offset值 = （当前页码 - 1）* 每页数量limit

@app.get("/book/get_book_list")
async def get_book_list(
    page: int = 1,
    page_size: int = 3,
    db: AsyncSession = Depends(get_database)
):
    # (页码 - 1) * 每页数量
    skip = (page - 1) * page_size

    # offset 跳过的记录数 ; limit 每页的记录数
    stmt = select(Book).offset(skip).limit(page_size)
    result = await db.execute(stmt)
    books = result.scalars().all()
    return books

既然前端就能执行分页功能了那为什么后端还需要执行分页？

答：

1. 数据量问题

• 前端分页的局限：如果后端一次性返回所有数据，当数据量很大（例如成千上万条）时，前端需要占用大量内存来存储数据，渲染大量DOM元素会导致页面卡顿，甚至崩溃。

• 后端分页的优势：后端只返回当前页所需的数据（例如 LIMIT 10 OFFSET 20），前端只需处理少量数据，内存和渲染压力极小，保证了流畅的用户体验。

2. 网络传输效率

• 一次性传输全部数据会消耗大量带宽，延长页面加载时间，尤其在移动网络环境下体验更差。

• 后端分页只传输当前页的数据，响应快，节省流量，页面能迅速呈现内容。

3. 数据库性能

• 后端可以利用数据库的索引和分页查询机制（如 LIMIT / OFFSET、游标分页等）高效地获取数据，避免将大量数据从数据库传到应用层再过滤。

• 如果前端分页，后端仍然需要查询全部数据，这不仅浪费数据库资源，还可能拖垮数据库。

4. 安全与权限

• 某些数据可能包含敏感信息（如用户隐私、商业机密），不应全部暴露给前端。后端分页可以在服务端控制数据访问权限，只返回用户有权查看的部分。

5. 用户体验与交互模式

• 常见的分页、无限滚动、加载更多等交互，都需要后端配合按需返回数据，而不是一次性加载全部。

• 例如搜索引擎、电商商品列表，用户通常只浏览前几页，后端分页可以按需加载，提升感知性能。

6. 接口通用性

• 后端提供分页接口后，可以被多个前端（Web、移动端、第三方）复用，每个客户端按需请求数据，无需各自实现完整数据加载逻辑。

增：

核心步骤：定义ORM对象 -> 添加对象到事务:add(对象) -> commit提交到数据库

class BookBase(BaseModel):
    id: int
    bookname:str
    author: str
    price:float
    publisher: str

@app.post("/book/add_book")
async def add_book(book: BookBase, db: AsyncSession = Depends(get_database)):
    # ORM对象 -> add -> commit
    book_obj = Book(**book.__dict__)  # 将用户提交的字典对象转换成ORM对象
    db.add(book_obj)
    await db.commit()
    return book

改：

核心步骤：查询get -> 属性重新赋值 -> commit提交到数据库

# 需求: 修改图书信息:先查再改
# 设计思路: 路径参数书籍id: 作用是查找; 请求体参数: 作用是新数据(书名、作者、价格、出版社)
class BookUpdate(BaseModel):
    bookname: str
    author: str
    price: float
    publisher: str


@app.put("/book/update_book/{book_id}")
async def update_book(book_id: int, data: BookUpdate, db: AsyncSession = Depends(get_database)):
    db_book = await db.get(Book, book_id)
    if db_book is None:
        raise HTTPException(
            status_code=404,
            detail="查无此书"
        )
    # 找到后则修改：重新赋值
    db_book.bokname = data.bookname
    db_book.author = data.author
    db_book.price = data.price
    db_book.publisher = data.publisher

    # 提交到数据库
    await db.commit()
    return db_book

删：

核心步骤：查询get -> delete删除 -> commit提交到数据库

@app.delete("/book/delete_book/{book_id}")
async def delete_book(book_id: int, db: AsyncSession = Depends(get_database)):
    # 先查再删 提交
    db_book = await db.get(Book, book_id)
    if db_book is None:
        raise HTTPException(
            status_code=404,
            detai="查无此书"
        )
    await db.delete(db_book)
    await db.commit()
    return {"msg": "删除图书成功"}