FastAPI 学习笔记

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，专为在 Python 中构建 RESTful API 而设计，异步支持，内置了await、async。

FastAPI 使用 Python 3.8+ 并基于标准的 Python 类型提示。

FastAPI 建立在 Starlette 和 Pydantic 之上，利用类型提示进行数据处理，并自动生成API文档。

Pydantic支持自动检验数据类型，避免需要手动判断。

FastAPI（API框架）本身只是框架，它运行uvicorn（服务器）和starlette（底层框架）上。

Client Request ↓ ASGI Server (Uvicorn) ↓ Event Loop (asyncio) ↓ FastAPI / Starlette Router ↓ 判断函数类型 ├── async def → Event Loop └── def → ThreadPool

starlette 底层框架

Starlette 可以理解为：

对原始 ASGI 的高级封装。

它提供：

• 路由系统

• Request / Response 对象

• 中间件

• WebSocket 支持

• 静态文件

• 后台任务

• 生命周期管理

• 模板支持

• 异步友好的 Web 基础设施 FastAPI 很多底层能力都来自 Starlette。

uvicorn 服务器

Uvicorn 不是业务框架，而是 运行应用的服务器。

它的核心职责：

• 监听 IP/端口

• 接收 HTTP / WebSocket 请求

• 解析请求协议

• 构造 ASGI scope

• 驱动 receive/send

• 调用你的 ASGI 应用

• 把应用返回的内容编码为 HTTP 响应发回客户端

• 管理事件循环、连接、并发、超时、日志

简单说：

Uvicorn = “把网络请求翻译成 ASGI 调用”的执行器。

FastAPI

FastAPI 是一个 面向接口开发 的高层框架，尤其适合：

• RESTful API

• 微服务

• 异步服务

• 带参数校验的后端接口

• 自动生成 OpenAPI 文档 它在 Starlette 之上增加了非常多开发效率功能（理解为在starlette的基础基础上进一步封装）：

• 路由声明更简洁

• 自动解析路径参数、查询参数、请求体

• 基于类型注解的数据校验

• 自动序列化返回值

• 依赖注入系统

• 自动生成 Swagger UI / ReDoc

• 更友好的异常处理

• 支持同步/异步函数混用

路由

路由定义基于Python的装饰器模式（Java注解）。

@app.get(“/”)

参数

路径参数

restful 位置：路径中，/book/{id}。路径参数传到了函数参数里面。

FastAPI 类型注解。Path，为参数申明提供额外的信息和校验。

导入FastAPI的Path函数。

查询参数

位置：URL？之后

作用：对资源集合进行过滤、排序、分页等操作

直接在函数参数里面取

Query()同样提供与Path差不多的信息和校验。

请求体参数

位置：HTTP请求的**消息体（Body）**中

作用：创建、更新资源，携带大量数据，如Json

HTTP协议中，一个完整的请求由三部分组成：请求行（方法、URL、协议版本）、请求头、（元数据信息（Content-Type、Authorization））、请求体（实际要发送的数据内容）

从请求头取数据：Header(…, alias=“Authorization”)

pydantic

自动校验参数类型

Filed 使用方法与传参时候的Path和Query注解。

from pydantic import BaseModel, Field
class User(BaseModel):
    username: str = Field(title="Username", description="Username", default="zhangsan")
    password: str

响应类型

FastAPI自动将路径的操作函数的返回的python对象（字典、列表、pydantic模型等）转换为Json格式，包装为JSONResponse返回，省去了手动序列化的步骤。

JSONResponse json格式 FileResponse 返回文件 StreamingResponse 返回流式响应 HTMLResponse 返回HTML内容 PlainResponse 返回纯文本 RedirectResponse 重定向

响应类型设置方式

装饰器中指定响应类：固定返回类型（HTML、纯文本等） 返回响应对象：文件、流式。

自定义响应格式

response_model 是路径操作装饰器的关键参数，通过一个pydantic的数据模型来约束返回的格式。

约定的返回值必须是这个格式，否则前端请求返回的 内部服务器报错（值少了报错，返回的值多了只取数据模型的值）

异常处理

对于客户端的引发的错误，可以使用fastapi.HTTPException 来终端正常处理流程

HTTPException(status_code=404, detail="Not found")

中间件

使用中间件为每个请求前后添加统一的处理逻辑（AOP，切面）

日志记录、身份认证、跨域处理、性能监控。。。

中间件是一个每次请求进入fastapi应用都会被执行的函数。（所有的都走中间件了）

顶部使用装饰器 @app.minddleware(“http”)

@app.middleware("http")
async def middleware(request: Request, call_next):
    print("请求前")
    response = await call_next(request)
    print("请求后")
    return response

多个中间件，执行顺序从下往上，因为上面的中间件也需要先执行下面的中间件，先下来再往上走

依赖注入系统

来共享通用逻辑，减少代码重复，

依赖项：可重复使用的组件（函数/类）

请求参数、业务逻辑、数据库连接、身份认证

创建依赖项，导入depends，申明依赖项

ORM（对象关系映射）

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

优势：减少重复的SQL代码，代码更简洁易读，自动处理数据连接和事务，自动防止SQL注入攻击

SQLAlchemy ORM：最灵活，企业级 Django ORM：封装好、上手快 Tortoise ORM：全异步，异步web服务、高并发API

初始化数据库连接引擎

定义基础模型类（对应表的对象都能用得到）

# 定义模型类，基类+表对应的模型类
class BaseModel(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now, comment="create time")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now, onupdate=func.now(),comment="update time")

对应表的对象（需要继承基础模型类）

class Book(BaseModel):
    __tablename__ = "book"
    id: Mapped[int] = mapped_column(primary_key=True, comment="book id")
    book_name: Mapped[str] = mapped_column(String(255), comment="book name")
    author: Mapped[str] = mapped_column(String(255), comment="book author")
    price: Mapped[float] = mapped_column(Float, comment="book price")
    publisher: Mapped[str] = mapped_column(String(255), comment="book publisher")

项目启动时创建

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

# @app.on_event("startup")
# async def startup():
#     await create_tables()

@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_tables()
    yield
    await async_engine.dispose()

app = FastAPI(lifespan=lifespan)

在路由中使用ORM

# 查询功能的结构，查询图书，orm的方式
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:
            await session.rollback() # 有异常回滚
            raise
        finally:
            await session.close()
@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_db)):

    result = await db.execute(select(Book))
    books = result.scalars().all()

    return books

数据库操作

1. 查询

核心：await db.execute(select(模型类))，返回一个ORM对象

获取所有数据 scalars().all()

获取单条数据：scalars().fitst();get(模型类, 主键值)

查询条件，select(Book).where(条件, 条件2, ////)

条件： 比较判断：== > < >= <=等 模糊查询：like() %（模糊匹配）， _ 匹配单个字符 与非查询：& ｜ ～ 包含查询：in_()

result.scalar_one_or_none()

聚合查询 聚合计算：func.方法(模型类.属性) count：统计行数量 avg：求平均值 max：求最大值 min：求最小值 sum：求和 分页查询 select.offset().limit() offset：跳过的记录数=（当前页码 - 1） * 每页数量limit limit：返回的记录数

2. 新增

定义ORM对象，添加对象到事务：add（对象） commit提交到数据库

class BookCreate(BaseModel):
    id: int
    book_name: str
    author: str
    price: float
    publisher: str

# 新增
@app.post("/book/add_book")
async def add_book(book: BookCreate, db: AsyncSession = Depends(get_db)):

    # book_obj = Book(**book.__dict__)
    book_obj = Book(**book.model_dump())
    db.add(book_obj)

    await db.commit()
    return book

3. 修改

查询get，属性重新赋值，commit提交到数据库 SQLAlchemy 会自动 追踪对象变化（dirty tracking）

@app.put('/book/update_book/{id}')
async def update_book(id: int, data:  UpdateBook,db: AsyncSession = Depends(get_db)):
    book_obj = await db.get(Book, id)
    if book_obj is None:
        raise HTTPException(status_code=404, detail="Book not found")
    book_obj.book_name = data.book_name
    book_obj.author = data.author
    book_obj.price = data.price
    book_obj.publisher = data.publisher
    await db.commit()
    return book_obj

4. 删除

查询get，delete删除，commit提交到数据库

async def delete_book(id: int, db: AsyncSession = Depends(get_db)):
    book_obj = await db.get(Book, id)
    if book_obj is None:
        raise HTTPException(status_code=404, detail="Book not found")
    await db.delete(book_obj)
    await db.commit()
    return book_obj

定义响应格式

成功响应

model_config = ConfigDict(from_attributes=True).

class UserInfoBase(BaseModel):
    """
    用户信息基础数据模型
    """    nickname: Optional[str] = Field(None, max_length=50, description="昵称")
    avatar: Optional[str] = Field(None, max_length=255, description="头像URL")
    gender: Optional[str] = Field(None, max_length=10, description="性别")
    bio: Optional[str] = Field(None, max_length=500, description="个人简介")

class UserInfoResponse(UserInfoBase):
    id: int
    username: str
    model_config = ConfigDict(
        populate_by_name=True,  # alias / 字段名 兼容使用
        from_attributes=True, # 允许从orm对象属性中取值
    )



class UserAuthResponse(BaseModel):
    token: str
    user_info: UserInfoResponse = Field(..., alias="userInfo"),

    model_config = ConfigDict(
        populate_by_name=True,  # alias / 字段名 兼容使用
        from_attributes=True, # 允许从orm对象属性中取值
    )

def success_response(message: str = "success", data = None):

    content = {
        "code": 200,
        "message": message,
        "data": data
    }

    return JSONResponse(content=jsonable_encoder(content))

异常响应

全局异常处理器，是注册在FastAPI应用级别的异常处理函数，用于捕获业务层、数据层以及系统层抛出的异常，并以统一的响应格式返回给前端。

SQL错误、外键关联失败

app.add_exception_handler(HTTPException, http_exception_handler)
app.add_exception_handler(IntegrityError, integrity_error_handler)
app.add_exception_handler(SQLAlchemyError, sqlalchemy_error_handler)
app.add_exception_handler(Exception, general_exception_handler)