Skip to content

API 设计与分页

接口写得多了,设计上的问题会冒出来:返回结构不统一(有的 {"data": ...} 有的直接返回)、错误格式不一样、列表接口数据一多前端卡死。这篇整理几个常见的设计规范:URL 命名、分页、过滤、统一响应格式、统一错误格式。

一、URL 命名

REST 风格的 URL 用名词复数,用 HTTP 方法表示动作:

方法URL作用
GET/articles列表
GET/articles/1单条
POST/articles创建
PATCH/articles/1部分更新
DELETE/articles/1删除

不用动词在 URL 里——/getArticles/createArticle/deleteArticle/1 这种写法不规范。动作交给 HTTP 方法表达,URL 只表达资源。

嵌套资源用层级 URL:/articles/1/comments 表示"文章 1 的评论"。

二、分页

列表接口数据量大时必须分页——一次返回几万条数据,前端渲染卡、网络传得慢、数据库查询也慢。

1 limit/offset 分页

最常见的方式:limit 每页多少条,skip(或 offset)跳过多少条。

python
@app.get("/articles")
def list_articles(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    # 限制 limit 最大值,防止前端传 limit=999999 拖垮数据库
    limit = min(limit, 100)

    articles = db.query(Article).offset(skip).limit(limit).all()
    total = db.query(Article).count()

    return {
        "items": [{"id": a.id, "title": a.title} for a in articles],
        "total": total,
        "skip": skip,
        "limit": limit,
    }

前端请求 /articles?skip=0&limit=20 拿第一页,skip=20 拿第二页。返回体里有 total(总共多少条),前端据此算总页数。

limit 一定要设上限(这里是 100)。不设的话前端传 limit=100000,数据库一下子查出十万条,内存和响应时间都会爆。

2 游标分页

limit/offset 在数据量大时有性能问题——offset=100000 意味着数据库要先扫过前十万条再取二十条。游标分页用"上一页最后一条的 ID"做起点:

python
@app.get("/articles")
def list_articles(after_id: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    limit = min(limit, 100)

    query = db.query(Article)
    if after_id > 0:
        query = query.filter(Article.id > after_id)

    articles = query.order_by(Article.id).limit(limit).all()

    return {
        "items": [{"id": a.id, "title": a.title} for a in articles],
        "next_cursor": articles[-1].id if len(articles) == limit else None,
    }

前端第一次请求 after_id=0,拿到 20 条,返回 next_cursor=20。第二次请求 after_id=20,从 ID 20 之后取下一页。游标分页不管翻到第几页,查询复杂度都一样——因为用的是 WHERE id > N,走索引直接定位。

游标分页的缺点是不能跳页(不能直接跳到第 50 页),只能"下一页""上一页"地翻。大部分列表接口不需要跳页,游标分页是更好的选择。

三、过滤和排序

1 过滤

过滤参数用查询参数,名字要语义清晰:

python
@app.get("/articles")
def list_articles(
    status: str = None,       # 按状态过滤
    keyword: str = None,      # 按标题搜索
    skip: int = 0,
    limit: int = 20,
    db: Session = Depends(get_db),
):
    query = db.query(Article)

    if status:
        query = query.filter(Article.status == status)

    if keyword:
        query = query.filter(Article.title.contains(keyword))

    articles = query.offset(skip).limit(limit).all()
    return {"items": [{"id": a.id, "title": a.title, "status": a.status} for a in articles]}

前端调 /articles?status=published&keyword=Python 拿到"已发布且标题包含 Python"的文章。每个过滤参数可选(默认 None),不传就不加那个条件。

2 排序

python
@app.get("/articles")
def list_articles(
    sort: str = "newest",     # newest(最新)或 oldest(最早)
    db: Session = Depends(get_db),
):
    query = db.query(Article)

    if sort == "oldest":
        query = query.order_by(Article.created_at.asc())
    else:
        query = query.order_by(Article.created_at.desc())

    return [{"id": a.id, "title": a.title} for a in query.all()]

四、统一响应格式

接口多了之后,每个接口返回格式不一致很烦——有的返回数组、有的返回对象、有的返回 {"data": ...}、有的直接返回。统一一个外壳格式:

python
def success(data, total=None):
    """统一成功响应。"""
    result = {"code": 0, "data": data}
    if total is not None:
        result["total"] = total
    return result


@app.get("/articles")
def list_articles(db: Session = Depends(get_db)):
    articles = db.query(Article).all()
    return success([{"id": a.id, "title": a.title} for a in articles])


@app.get("/articles/{article_id}")
def get_article(article_id: int, db: Session = Depends(get_db)):
    article = db.query(Article).filter(Article.id == article_id).first()
    if not article:
        raise HTTPException(status_code=404, detail="文章不存在")
    return success({"id": article.id, "title": article.title})

所有成功响应都是 {"code": 0, "data": ...},前端封装的 request 函数里统一判断 code === 0这个外壳格式不是强制的,团队约定好就行——关键是统一,格式本身怎么定影响不大。

五、统一错误处理

分散在各接口里的 raise HTTPException 不便于统一管理。FastAPI 支持全局异常处理器:

python
from fastapi import Request
from fastapi.responses import JSONResponse


# 自定义业务异常
class BusinessError(Exception):
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message


# 全局异常处理
@app.exception_handler(BusinessError)
async def business_error_handler(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=200,
        content={"code": exc.code, "message": exc.message},
    )


@app.exception_handler(Exception)
async def general_error_handler(request: Request, exc: Exception):
    # 兜底:未预期的异常统一返回 500,不暴露堆栈
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "服务器内部错误"},
    )

业务逻辑里抛 BusinessError,全局处理器统一转成固定格式。未预期的异常被兜底处理器接住,不会把 Python 堆栈泄露给前端——堆栈信息只在服务端日志里看到。

接口里用:

python
@app.post("/articles")
def create_article(article: ArticleCreate, db: Session = Depends(get_db)):
    if len(article.title) < 2:
        raise BusinessError(code=1001, message="标题太短")

    new_article = Article(title=article.title, content=article.content)
    db.add(new_article)
    db.commit()
    return success({"id": new_article.id})