FastAPI

简介

异步，高性能
Pydantic 类型提示与验证
可自动生成 API 可交互式文档

创建项目

Python

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

if __name__ == "__main__":
    # 直接在代码中启动uvicorn服务器
    uvicorn.run(
        app="main:app",
        host="0.0.0.0",
        port=8000,
        reload=True 
    )

Swagger文档：/docs

路由

Python

@app.get("/")

app：实例对象

get：请求方法

路由分发

使用 APIRouter 把路由使用注册的方式拆分，方便维护。

python

# 子路由
from fastapi import APIRouter

router = APIRouter(
    prefix="/shop",
    tags=["购物模块"]  # 用于文档分类
)


@router.get("/")
def get_shops():
    return {"message": "all shops"}

python

# 挂载到 app 实例
app.include_router(router=shop.router)

传参

路径参数

Python

@app.get("/user/{id}")
async def read_items(id: int, username: str):
    return {"id": id, "username": username}

请求体参数

定义类型
类型注解

Python

class User(BaseModel):
    id: int
    username: str

@app.post("/user")
async def add_user(user: User):
    user_list.append(user)
    return user

传参校验

Path、Query

Python

from fastapi import Path, Query

@app.get("/user/{id}")
async def get_user(id: int = Path(..., le=10), username: str =  Query(..., max_length=20)):  # ... 表示必填项
    return {"id": id, "username": username}

Field

这个跟上面俩不在一个包

Python

from pydantic import BaseModel, Field

class User(BaseModel):
    id: int = Field(..., le=999)
    username: str = Field(..., max_length=10)

@app.post("/user")
async def add_user(user: User):
    user_list.append(user)
    return user

响应类型

FastAPI 默认会将 Python 对象转换成 JSON

在装饰器指定响应类型，用于HTML

Python

@app.get("/", response_class=HTMLResponse)
async def root():
    return "<h2> Hello World </h2>"

返回响应对象，用于文件下载、流式响应

Python

@app.get("/file")
async def file():
    file_path = "icon.jpg"
    # 仅支持本地路径
    return FileResponse(file_path)

自定义响应数据格式：用于约束API返回值

Python

class User(BaseModel):
    id: int = Field(..., le=999)
    username: str = Field(..., max_length=10)

@app.get("/user_demo", response_model=User)
async def user_demo():
    return {
        "id": 123,
        "username": "wmh",
        "class": "CS"   # 多的会直接过滤掉，少了(必填的)会500
    }

异常处理

fastapi.HTTPException

Python

@app.get("/user/{id}")
async def get_user(id: int):
    for user in user_list:
        if user.id == id:
            return user
    # raise 类似于 throw
    raise fastapi.HTTPException(status_code=404, detail="User not found")

请求头

Python

@app.get("/items/")
def read_item(user_agent: str = Header(None), session_token: str = Cookie(None)):    # None 是默认值
    return {"User-Agent": user_agent, "Session-Token": session_token}

重定向

Python

@app.get("/redirect")
def redirect():
    return RedirectResponse(url="/")

依赖注入

中间件会控制所有请求
依赖注入控制指定接口

依赖是一个函数，它可以使用与路径操作函数相同的参数（查询参数、路径参数、请求体等）

Python

# 1. 定义依赖函数
def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}


# 2. 在路由中使用依赖
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    # commons 接收依赖函数的返回值
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    # 多个路由可以复用同一个依赖
    return commons

除了函数，也可以用类作为依赖

Python

# 用类声明依赖
class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


# 使用类作为依赖
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

Annotated 的语义本质是：在保持类型不变的前提下，为类型添加额外的"说明书"，

在 FastAPI 中，这就是：Annotated[类型, 来源/验证规则] = "这个参数是某类型，请按指定方式从请求中提取并验证它"。

Python

# Python: 通过 Annotated 附加元数据
x: Annotated[int, "单位: 米"]

# TypeScript: 通过装饰器或 JSDoc
/** @param {number} x - 单位: 米 */
function foo(x: number) {}

# Java: 通过注解
@Param(description = "单位: 米")
int x

只需要依赖的副作用（如权限校验），不需要其返回值。可以直接在装饰器的 dependencies 参数中声明

Python

# 依赖：校验 API Key
async def verify_api_key(x_api_key: str = Header()):
    if x_api_key != "secret-key":
        raise HTTPException(status_code=400, detail="X-API-Key invalid")


# 在装饰器中使用依赖，不需要返回值
@app.get("/items/", dependencies=[Depends(verify_api_key)])
async def read_items():
    return [{"item": "Foo"}]

可以放在实例上，全局生效

Python

async def verify_token(x_token: str = Header()):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


# 全局依赖：所有路由都需要通过 token 校验
app = FastAPI(dependencies=[Depends(verify_token)])

当依赖需要执行清理操作（如关闭数据库连接）时，使用 yield

Python

# 使用 yield 的依赖：请求前创建连接，请求后关闭连接
def get_db():
    db = "database_connection"  # 模拟创建数据库连接
    try:
        yield db  # 请求处理期间提供数据库连接
    finally:
        print("关闭数据库连接")  # 请求完成后清理资源


@app.get("/items/")
async def read_items(db: str = Depends(get_db)):
    return {"db": db}

表单数据

默认请求头传参是 JSON（使用Pydantic 模型的 BaseModel）

Python

@app.post("/login/")
async def login(
    username: str = Form(),      # 必填表单字段
    password: str = Form(None),      # 可选表单字段
):
    return {"username": username}

中间件（AOP/拦截器）

Python

@app.middleware("http")
async def middleware(request, call_next):
    print("middleware1 start") # 1
    response = await call_next(request)
    print("middleware1 end") # 4
    return response


@app.middleware("http")
async def middleware2(request, call_next):
    print("middleware2 start") # 2
    response = await call_next(request)
    print("middleware2 end") # 3
    return response
middleware2 start
middleware1 start
middleware1 end
middleware2 end
# 请求 -> 中间件A(前) -> 中间件B(前) -> 路由函数
# 响应 <- 中间件A(后) <- 中间件B(后) <- 路由函数

请求日志中间件

Python

import logging

logger = logging.getLogger("uvicorn.error")


@app.middleware("http")
async def log_requests(request: Request, call_next):
    # 记录请求信息
    logger.info(f"请求: {request.method} {request.url}")

    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time

    # 记录响应信息
    logger.info(
        f"响应: {request.method} {request.url} "
        f"状态码={response.status_code} 耗时={process_time:.3f}s"
    )

    return response

强制 HTTPS 重定向，FastAPI 继承自 Starlette，可以直接使用 Starlette 提供的中间件

Python

from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

# 强制 HTTPS 重定向
app.add_middleware(HTTPSRedirectMiddleware)

GZip 压缩中间件

Python

app = FastAPI()

# 当响应大小超过 1000 字节时自动压缩
app.add_middleware(GZipMiddleware, minimum_size=1000)

处理跨域

Python

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 开发环境：允许所有源（仅用于开发！）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],         # 允许所有源
    allow_credentials=True,
    allow_methods=["*"],         # 允许所有方法
    allow_headers=["*"],         # 允许所有请求头
)

静态文件

使用 Starlette 的 StaticFiles 实现，支持 Jinja2 模板

Bash

pip install aiofiles    # StaticFiles 包含在 aiofiles 中
# 静态文件
app.mount(
    "/static",    # URL路径前缀，如访问 /static/logo.png
    StaticFiles(directory="static"),    # 本地目录名，存放静态文件的文件夹
    name="static"    # 内部使用的名称
)

# Jinja2 模板
templates = Jinja2Templates(
    directory="templates"
)

@app.get("/index")
async def index(request: fastapi.Request, name: str):
    return template.TemplateResponse(
        name="index.html",     # 需要传递 request 对象，因为模板中可能需要使用请求上下文
        request=request,
        context={
            "name": name
        }
    )

index.html

HTML

<h1>Hello, {{ name }}!</h1>

数据库集成 ORM

安装依赖

pip install sqlalchemy

1、database.py：配置数据库连接

python

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

MYSQL_URL = "mysql+pymysql://root:123.com@localhost:3306/fastapi_db"

# 引擎
engine = create_engine(MYSQL_URL, echo=True)

# 会话工厂
SessionLocal = sessionmaker(
    autocommit=False,  # 关闭自动提交，需手动 commit()
    bind=engine
)

# 基类
class Base(DeclarativeBase):
    pass

2、models.py：定义实体类

python

from sqlalchemy import Column, Integer, String

from database import Base

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255))
    password = Column(String(255))

3、schemas.py：定义 API 数据封装类（DTO、VO）

python

from pydantic import BaseModel


class UserBase(BaseModel):
    name: str

class UserCreate(UserBase):
    password: str

class UserVO(UserBase):
    id: int

class UserUpdate(UserBase):
    id: int
    password: str

这里的 UserUpdate 的字段虽然和 User 相同，但是他们继承了不同的父类。User 没有集成 BaseModel，不能作为 API 的输入输出。这里与 Java 的实体类不同。

4、crud.py：数据库操作（相当于 Spring 的 DAO 层）

python

from sqlalchemy.orm import Session

from models import User
from schemas import UserCreate, UserUpdate


def create_user(db: Session, user: UserCreate):
    new_user = User(name=user.name)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

def get_user(db: Session, id: int):
    return db.query(User).filter(User.id == id).first()

def delete_user(db: Session, id: int):
    user = db.query(User).filter(User.id == id).first()
    db.delete(user)
    db.commit()
    return

def update_user(db: Session, new_user: UserUpdate):
    user = db.query(User).filter(User.id == new_user.id).first()
    user.name = new_user.name
    user.password = new_user.password
    db.commit()
    db.refresh(user)
    return user

5、main.py

python

import fastapi
import uvicorn
from fastapi import Depends
from sqlalchemy.orm import Session

import schemas
from database import SessionLocal
import crud

app = fastapi.FastAPI()

# 依赖：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/user/{id}", response_model=schemas.UserVO)
def get_user(id: int, db: Session = Depends(get_db)):
    return crud.get_user(db, id)


@app.post("/user", response_model=schemas.UserVO)
def add_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    return crud.create_user(db, user)


@app.delete("/user/{id}")
def delete_user(id: int, db: Session = Depends(get_db)):
    crud.delete_user(db, id)
    return {"msg": "ok"}


@app.put("/user", response_model=schemas.UserVO)
def update_user(user: schemas.UserUpdate, db: Session = Depends(get_db)):
    return crud.update_user(db, user)


if __name__ == "__main__":
    uvicorn.run(
        app="main:app",
        host="127.0.0.1",
        port=8000,
        reload=True
    )

FastAPI：异步 Web 框架与接口开发https://wmhwiki.cn/posts/2026/fastapi

作者wmh

发布于9/8

更新于今日内

许可协议 CC BY-NC-SA 4.0

署名-非商业性使用-相同方式共享 4.0 国际

后端技术 FastAPI

FastAPI：异步 Web 框架与接口开发

FastAPI ​

简介 ​

创建项目 ​

路由 ​

路由分发 ​

传参 ​

传参校验 ​

响应类型 ​

异常处理 ​

请求头 ​

重定向 ​

依赖注入 ​

表单数据 ​

中间件（AOP/拦截器） ​

静态文件 ​

数据库集成 ORM ​

FastAPI

简介

创建项目

路由

路由分发

传参

传参校验

响应类型

异常处理

请求头

重定向

依赖注入

表单数据

中间件（AOP/拦截器）

静态文件

数据库集成 ORM