FastAPI 是一个用于构建 API 的现代、快速（高性能）的 Web 框架，使用 Python 并基于标准的 Python 类型提示。本篇文章带你入门FastApi。

快速上手

安装依赖

shell
pip install "fastapi[standard]"

入口文件

python
from fastapi import FastAPI

# 创建 FastApi 应用实例
app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

pycharm 编辑配置

接口文档

浏览器访问 http://127.0.0.1:8000/docs 或者 http://127.0.0.1:8000/redoc

处理请求

请求方法

基于RESTful的装饰器写法

@app.get()
@app.post()
@app.put()
@app.delete()
@app.options()
@app.head()
@app.patch()
@app.trace()

路径参数

python
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

复杂数据校验

普通写法

python
from fastapi import FastAPI, Path

# item_id 必须在100 - 1000 之间
# Path(...,description="item_id 必须在 100 至 1000 之间", gt=100,lt= 1000)
# ... 表示没有默认值，路径参数 不能有默认值，所以这里必须写...
@app.get("/items/{item_id}")
def read_item(item_id: int = Path(...,description="item_id 必须在 100 至 1000 之间", gt=100,lt= 1000), q: str | None = None):
    return {"item_id": item_id, "q": q}

元注解写法

python
from typing import Annotated
from fastapi import FastAPI, Path

@app.get("/items2/{item_id}")
def read_item2(item_id: Annotated[int,Path(...,gt=100,lt=1000)], q: str | None = None):
    return {"item_id": item_id, "q": q}

推荐使用元注解写法，该写法可以写很多校验信息，灵活

查询参数

python
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

可选参数

python
@app.get("/books/list")
def get_books(page: int = 1,pageSize: int = 10):

    return {
        "total": 84,
        "pageNum": page,
        "pageSize": pageSize
    }

如果查询参数有默认值，该参数就是可选的

使用 Query() 指定复杂参数校验

python
from typing import Annotated

from fastapi import FastAPI, Path, Query

@app.get("/books/list")
def get_books(page_size: Annotated[int, Query(..., description="每页数量", gt=10, lt=1000)],
              page: int = Query(1, description="页码", gt=0, lt=100), ):
    return {
        "total": 84,
        "pageNum": page,
        "pageSize": page_size
    }

请求头

python
@router.post("/items/{item_id}")
def read_items(
        item_id,
        q,
        user_agent: str | None = Header(None, description="用户代理浏览器")
):
    return {
        "code": 200,
        "data": {
            "q": q,
            "user_agent": user_agent
        }
    }

Header比Path、Query和Cookie提供了更多功能。大部分标准请求头用连字符分割，即减号 - 。但user-agent 这样的变量在python 中是无效的。默认情况下：

Header 把参数名中的字符下划线_改为连字符-来提取并存档请求头。
HTTP 的请求头不区分大小写，可以使用python标准样式(snake_case)进行声明。
可以像在python代码中一样使用 user_agent
如需禁用下划线自动转换为连字符，可以把Header的convert_underscores参数设置为False

请求体

python
from typing import Annotated
from fastapi import APIRouter, Header, Body
from pydantic import BaseModel

class Book(BaseModel):
    title: str
    price: float
    author: str


@router.post("/items/body")
def read_body(item: Annotated[dict, Body(..., description="请求体参数")]):
    return item


@router.post("/items/books")
# 自动从请求体中获取数据
def read_body(item: Book):
    return item

表单数据

python
from typing import Annotated

from fastapi import APIRouter, Header, Body, Form, File, UploadFile
from pydantic import BaseModel, Field

@router.post("/form")
def read_form(username: Annotated[str, Form(..., description="用户名")],
              password: Annotated[str, Form(..., description="密码")]):
    return {
        "username": username,
        "password": password,
    }


class User(BaseModel):
    username: str = Field(description="用户名")
    password: str = Field(description="密码")


@router.post("/form2")
def read_form2(user: Annotated[User, Form(..., description="用户信息")]):
    return user


# 获取上传的文件
## 1. 第一种写法，如果遇到大文件，字节流会把服务器撑爆
@router.post("/upload")
def upload_file(file: bytes = File(...)):
    return {
        "file_size": len(file)
    }


# 第2种写法
@router.post("/upload2")
async def upload_file2(file: UploadFile):
    # UploadFile 对文件的读写操作都是异步的
    # 保存文件
    contents = await file.read()
    file_size = len(contents)
    with open(f"/upload/{file.filename}", "wb") as f:
        f.write(contents)

    return {
        "file_name": file.filename,
        "content_type": file.content_type,
        "file_size": file_size
    }

在文件上传案例中，推荐使用第二种写法，使用UploadFile

与 bytes 相比，使用 UploadFile 有多项优势：

使用 scoped缓冲写入机制：
- 文件会先存储在内存中，直到达到最大上限，超过改上限会写入磁盘
- 超出内存阈值限制，文件会被放到一个临时位置
适合处理图像、视频、大型二进制文件
可以获取上传文件的元数据

其它数据

python
@router.get("/items/x/other")
# 从Request中获取数据
def read_body(req: Request):
    return {
        "url": req.url,
        "method": req.method,
        "cookie": req.cookies,
        "user_agent": req.headers
    }

pydantic 基础入门

快速上手

python
from datetime import datetime

from pydantic import BaseModel, PositiveInt


class User(BaseModel):
    id: int  # 必填 int
    name: str = "默认名字"  # 可选，带默认值
    signup_ts: datetime | None  # 可选，可以为None
    scores: dict[str, PositiveInt]  # value必须是正整数


# 原始外部数据（可能类型混乱）
raw_data = {
    "id": "1001",  # 字符串自动转int
    "signup_ts": "2026-08-22 10:00:00",  # 字符串自动转datetime
    "scores": {"math": "90", "english": 85}
}

# 解析+校验
user = User(**raw_data)
print(user.id, type(user.id))
print(user.model_dump())  # 转字典
print(user.model_dump_json())  # 转json字符串

from pydantic import ValidationError

try:
    User(id=-1, signup_ts=None, scores={"math": -5})
except ValidationError as e:
    print(e.errors()) # 查看所有错误详情

字段约束：Field

Field 用来设置长度、大小、正则、描述、示例等规则常用参数：

gt/ge/lt/le：大于、大于等于、小于、小于等于
min_length / max_length：字符串长度
pattern：正则匹配
default 默认值；default_factory 动态默认（列表 / 字典必用！避免共享引用）
description、examples：生成 JSON 文档用

python
from pydantic import BaseModel, Field, EmailStr

class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=20, pattern=r"^[a-zA-Z0-9_]+$")
    email: EmailStr # 内置邮箱格式校验
    age: int = Field(ge=0, le=120, description="年龄 0~120")
    tags: list[str] = Field(default_factory=list) # ✅ 每次实例新建空列表（不要直接=[]）

❌ 坑：tags: list[str] = [] 多个实例会共用同一个列表！必须 default_factory=list

内置常用校验类型

python
from pydantic import EmailStr, HttpUrl, PositiveInt, PastDate, FutureDate
from uuid import UUID

class Demo(BaseModel):
    email: EmailStr
    website: HttpUrl
    price: PositiveInt
    create_time: PastDate # 必须是过去日期
    uid: UUID

嵌套模型（复杂结构化数据）

python
from pydantic import BaseModel

class Address(BaseModel):
    province: str
    city: str

class User(BaseModel):
    name: str
    addr: Address       # 嵌套单个模型
    addr_list: list[Address] # 嵌套模型数组

data = {
    "name": "张三",
    "addr": {"province": "河北", "city": "廊坊"},
    "addr_list": [{"province": "北京", "city": "北京"}]
}
u = User(**data)
print(u.addr.city)

自定义校验器

@field_validator 单字段校验（V2 推荐）替代 v1 的 @validator，必须加 @classmethod

python
from pydantic import BaseModel, field_validator

class User(BaseModel):
    username: str
    password: str

    @field_validator("username")
    @classmethod
    def check_name(cls, v: str):
        if v in ["admin", "root"]:
            raise ValueError("用户名禁止使用admin/root")
        return v.strip() # 清洗数据，返回处理后的值

    @field_validator("password")
    @classmethod
    def check_pwd(cls, v: str):
        if len(v) < 8:
            raise ValueError("密码至少8位")
        return v

@model_validator 模型级校验（多字段联动） mode="before"：校验前执行；mode="after"：所有字段校验完成后执行适合：两次密码对比、条件必填等场景

python
from pydantic import BaseModel, model_validator
from typing import Self

class Register(BaseModel):
    pwd: str
    confirm_pwd: str

    @model_validator(mode="after")
    def check_pwd_eq(self) -> Self:
        if self.pwd != self.confirm_pwd:
            raise ValueError("两次密码不一致")
        return self

模型配置 model_config（ConfigDict）

V2 不再使用内部 class Config，改用 model_config = ConfigDict()

python
from pydantic import BaseModel, ConfigDict

class User(BaseModel):
    name: str
    age: int

    model_config = ConfigDict(
        extra="ignore",        # ignore:忽略多余字段；forbid:禁止多余字段；allow:允许
        from_attributes=True, # 支持ORM对象直接转模型（v1 orm_mode=True）
        populate_by_name=True, # 支持别名赋值
        frozen=False,          # True 不可变模型，实例创建后不能修改
    )

序列化 & 反序列化完整 API

python
class User(BaseModel):
    id: int
    name: str

# 1. dict 构造实例
u1 = User(id=1, name="Alice")
u2 = User(**{"id":2, "name":"Bob"})

# 2. 通用校验构造（推荐，兼容dict/ORM对象）
u3 = User.model_validate({"id":3, "name":"Charlie"})

# 3. JSON字符串解析
json_str = '{"id":4,"name":"David"}'
u4 = User.model_validate_json(json_str)

# 4. 序列化
print(u1.model_dump()) # dict
print(u1.model_dump_json(indent=2)) # json字符串

# 序列化控制：排除字段、只包含部分字段
u1.model_dump(exclude={"id"})
u1.model_dump(include={"name"})
u1.model_dump(exclude_unset=True) # 只保留传入赋值的字段（更新接口常用）

枚举、字面量 Literal

python
from enum import Enum
from pydantic import BaseModel
from typing import Literal

class Role(str, Enum):
    ADMIN = "admin"
    USER = "user"

class User(BaseModel):
    role: Role
    status: Literal["enable", "disable"] # 只能二选一

常用实战场景

场景 1：ORM 对象转模型（SQLAlchemy 配合使用）

python
from pydantic import BaseModel, ConfigDict

class UserResp(BaseModel):
    id: int
    name: str
    model_config = ConfigDict(from_attributes=True)

# orm_user 是 SQLAlchemy 查询出来的ORM实例
resp = UserResp.model_validate(orm_user)

响应处理

默认响应内容机制

如果试图函数返回的是dict/list/pydantic模型：fastApi默认会把它转换成application/json响应(JSONResponse)
如果返回的是str:默认响应类型是 PlainTextResponse
如果返回的是 bytes：默认响应类型是Response（application.octet-stream）

response_model

基本用法

python
from fastapi import APIRouter
from fastapi.responses import HTMLResponse

router = APIRouter(prefix="/resp", tags=["响应处理"])


@router.get("/")
async def read_items():
    return HTMLResponse(
        content="<h1>你好，fastApi</h1>",
        status_code=200,
        headers={"X-Custom": "demo"}
    )

response_model=XXX：声明 JSON 返回用哪个 Pydantic 模型做数据校验，只用于返回数据是json 的接口

常用Response类型

响应状态吗 fastapi.status 封装了常用状态码


rom fastapi import APIRouter, status
from fastapi.responses import HTMLResponse
from pydantic import BaseModel

@router.get("/items/{item_id}", response_model=Item, status_code=status.HTTP_201_CREATED)
async def read_item(item_id: int):
    return {
        "name": "Pen",
        "price": 1.5,
    }

JSONResponse

python
@router.get("items/x/x1")
async def read_item():
    return JSONResponse(
        status_code=status.HTTP_201_CREATED,
        content={
            "name": "PEN",
            "price": 1.5
        }
    )

PlainTextResponse
HTMLResponse
RedirectResponse
StreamResponse
FileResponse:文件下载
UJSONResponse
ORJSONResponse
自定义响应类型

python
from fastapi import APIRouter, status, Response
from fastapi.responses import HTMLResponse, JSONResponse
from pydantic import BaseModel

class XMLResponse(Response):
    media_type = "application/xml"
    
@router.get("items/x/xml")
async def read_item():
    return XMLResponse(content="<note><to>Tove<to><from>Jain</from></note>")

依赖注入

依赖注入是一种将函数或类所需的"外部资源"通过参数传入的技术，而不是在函数内部自行创建资源，这样可以提高可测试性与复用性。在FastApi中依赖注入常用于：

共用的数据库连接
用户认证信息
配置参数
权限检查
复用逻辑

基本用法

FastApi 通过 Depends 来实现依赖注入

抽离公共方法

python
from typing import Annotated

from fastapi import Header, Query

from core.exceptions import BizException


async def get_token(x_token: str | None = Header(default=None)):
    # 示例全局依赖：登录校验
    if not x_token:
        raise BizException(401, "未登录")
    return x_token


def pagination(page_num: Annotated[int, Query(description="页码", gt=0)],
               page_size: Annotated[int, Query(description="每页数量", gt=0, lt=100)] = 10,
               q: Annotated[str, Query(description="查询字符串")] = None):
    return {
        "page": page_num,
        "page_size": page_size,
        "q": q
    }

在需要的地方通过Depends注入

python
from fastapi import APIRouter, Depends

from core import pagination, get_token

router = APIRouter(prefix="/dep", tags=["依赖注入"])


@router.get("/books")
def get_books(page_param: dict = Depends(pagination)):
    return {
        "books": ["book1", "book2"],
        **page_param
    }


@router.post("/books/add")
def add_user(
        token: str = Depends(get_token)
):
    return {
        "code": 200,
        "data": {},
        "message": "数据添加成功"
    }

带资源清理的依赖(yield)

某些依赖需要在使用后释放资源，比如数据库连接，可以使用yield

python
from core.config import settings
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine(
    settings.DATABASE_URL,
    connect_args={"check_same_thread": False} if "sqlite" in settings.DATABASE_URL else {}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

python
from fastapi import APIRouter, Depends

from core import pagination, get_token, get_db

@router.get("/books/list")
def read_books(db=Depends(get_db)):
    return db.execute("select * from books").fetchall()

异常处理

在FastAPI中，错误(异常)处理机制是基于异常捕获(exception handlers)的，它允许你优雅地处理各种类型的错误(HTTP错误、自定义业务异常、验证错误等)并统一返回格式化的响应。

HTTPException

直接抛出HTTPExcepton，FastAPI会自动处理，返回指定状态码和内容的错误响应

python
from fastapi import APIRouter, HTTPException, status

router = APIRouter(prefix="/exce", tags=["异常处理"])


@router.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="未找到该数据")
    return {
        "item_id": item_id
    }

响应内容如下，响应码404

json
{
  "detail": "未找到该数据"
}

在真实业务场景中，一般都是自定义错误响应，响应码为200，响应内容里面还有一个code,前端根据里面的code做具体区分

自定义错误响应

封装统一的异常处理器

python
from fastapi import Request, FastAPI
from fastapi.responses import JSONResponse


class BizException(Exception):
    def __init__(self, code: int, msg: str):
        self.code = code
        self.msg = msg


def register_exception_handler(app: FastAPI):
    @app.exception_handler(BizException)
    async def biz_exception_handler(request: Request, exc: BizException):
        return JSONResponse(status_code=200, content={
            "code": exc.code,
            "msg": exc.msg,
            "data": None
        })

    @app.exception_handler(Exception)
    async def global_exception_handler(request: Request, exc: Exception):
        return JSONResponse(status_code=500, content={
            "code": 500,
            "msg": f"服务器异常: {str(exc)}",
            "data": None
        })

在main.py中应用


app = FastAPI()
register_exception_handler(app)

python

@router.get("/items2/{item_id}")
def read_item2(item_id: int):
    if item_id == 3:
        raise BizException(code=0, msg="业务异常")
    return {
        "item_id": item_id
    }

响应内容如下

json
{
  "code": 0,
  "msg": "业务异常",
  "data": null
}

统一封装返回消息

python
from enum import Enum

class HttpCode(str, Enum):
    """HTTP基础业务状态码"""
    SUCCESS = "success"  # 成功状态
    FAIL = "fail"  # 失败状态
    UNAUTHORIZED = "unauthorized"  # 未授权
    NOT_FOUND = "not_found"  # 未找到
    FORBIDDEN = "forbidden"  # 无权限
    VALIDATE_ERROR = "validate_error"  # 数据验证错误
    
    
    
from typing import Any

from fastapi import status
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field

from pkg.http_code import HttpCode


class RespModel(BaseModel):
    code: HttpCode = HttpCode.SUCCESS
    message: str = Field(default="")
    data: Any = Field(default_factory=dict)


def success_json(data: Any = None):
    """成功数据响应"""
    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content=RespModel(data=data).model_dump()
    )


def fail_json(data: Any = None):
    """失败数据响应"""
    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content=RespModel(data=data, code=HttpCode.FAIL).model_dump()
    )


def validate_error_json(errors: dict = None):
    """数据校验错误响应"""
    first_key = next(iter(errors))
    if first_key is not None:
        msg = errors[first_key][0]
    else:
        msg = "数据验证错误"

    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content=RespModel(
            code=HttpCode.VALIDATE_ERROR,
            data=errors,
            message=msg
        ).model_dump()
    )


def message(code: HttpCode = None, msg: str = ""):
    """基础消息响应"""
    return JSONResponse(
        status_code=status.HTTP_200_OK,
        content=RespModel(
            code=code,
            data={},
            message=msg
        ).model_dump()
    )


def success_message(msg: str = ""):
    """成功消息响应"""
    return message(code=HttpCode.SUCCESS, msg=msg)


def fail_message(msg: str = ""):
    """失败消息响应"""
    return message(code=HttpCode.FAIL, msg=msg)


def not_found_message(msg: str = ""):
    """未找到消息响应"""
    return message(code=HttpCode.NOT_FOUND, msg=msg)


def unauthorized_message(msg: str = ""):
    """未授权消息响应"""
    return message(code=HttpCode.UNAUTHORIZED, msg=msg)


def forbidden_message(msg: str = ""):
    """无权限消息响应"""
    return message(code=HttpCode.FORBIDDEN, msg=msg)

python
@router.post("/user/login")
def login(username: Annotated[str, Form(description="用户名")], password: Annotated[str, Form(description="密码")]):
    if username == "admin" and password == "123456":
        return success_json({
            "token": "xxxxx"
        })
    return fail_message("用户名或密码错误")

中间件

在FastAPI中，中间件(middleware)是实现全局请求/响应处理逻辑的核心机制之一，可以在进入路由前或返回后执行一些通用逻辑，比如：

记录访问日志
计算接口耗时
统一添加/修改响应头
进行请求身份检查或自定义限流
捕获未处理的异常(比 exception_handler 更底层)

FastAPI 中间件 = 全局请求/响应拦截器

log_middleware.py

python
from fastapi import Request


async def log_middleware(req: Request, call_next):
    print("请求开始：", req.url)

    # 在调用目标方法之前可以做额外操作，例如 鉴权

    # 调用目标方法
    resp = await call_next(req)

    print("响应结束：", resp.status_code)
    # 对响应做额外修改
    return resp

customer_middleware.py

python
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint


class CustomHeaderMiddleware(BaseHTTPMiddleware):
    """自定义中间件"""

    async def dispatch(self, req: Request, call_next: RequestResponseEndpoint) -> Response:
        print("自定义中间件 CustomHeaderMiddleware:")
        print("请求开始：", req.url)

        # 在调用目标方法之前可以做额外操作，例如 鉴权

        # 调用目标方法
        resp = await call_next(req)

        print("响应结束：", resp.status_code)
        # 对响应做额外修改
        # 统一追加响应头
        resp.headers["X-Server"] = "FastAPI-Demo"
        return resp

register_middleware.py

python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from .custom_middleware import CustomHeaderMiddleware
from .log_middleware import log_middleware


def register_middleware(app: FastAPI):
    # 注册中间件
    app.middleware("http")(log_middleware)
    # app.middleware("http")(CustomHeaderMiddleware)
    app.add_middleware(CustomHeaderMiddleware)
    
    # 最后注册，before前置调用是最先执行
    app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_credentials=True,
                       allow_headers=["*"])

在main.py中调用注册函数

python
app = FastAPI()
register_middleware(app)

注册顺序是 log_middleware -》 CustomHeaderMiddleware

before 前置调用顺序是按照注册顺序的反向顺序 after 后置调用按照注册顺序的顺序

声明周期

事件回掉函数

start_up: 应用启动 shutdown: 应用停止


# 这种写法在新版本被弃用
@app.on_event("startup")
def start_up():
    print("startup...")
  
@app.on_event("shutdown")
def shutdown():
    print("shutdown")

@app.on_event("startup") / @app.on_event("shutdown") 在新版 FastAPI（≥0.96）已经标记废弃，官方统一改用 lifespan 异步上下文管理器管理应用生命周期

新版写法

python
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ========== yield 之前 = startup 启动逻辑 ==========
    print("🚀 应用启动，初始化资源（数据库连接池、Redis、加载模型）")

    yield  # 交出控制权，服务正式开始接收请求

    # ========== yield 之后 = shutdown 关闭清理 ==========
    print("🛑 应用关闭，释放资源（关闭连接池）")

# 实例化时传入 lifespan
app = FastAPI(lifespan=lifespan)

目录

快速上手

安装依赖

入口文件

pycharm 编辑配置

接口文档

处理请求

请求方法

路径参数

查询参数

使用 Query() 指定复杂参数校验

请求头

请求体

表单数据

其它数据

pydantic 基础入门

快速上手

字段约束：Field

内置常用校验类型

嵌套模型（复杂结构化数据）

自定义校验器

模型配置 model_config（ConfigDict）

序列化 & 反序列化完整 API

枚举、字面量 Literal

常用实战场景

响应处理

默认响应内容机制

response_model

基本用法

常用Response类型

依赖注入

基本用法

带资源清理的依赖(yield)

异常处理

HTTPException

自定义错误响应

统一封装返回消息

中间件

声明周期

事件回掉函数