尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

Python后端RESTful API设计规范:从资源建模到工程落地

Python后端RESTful API设计规范:从资源建模到工程落地 我见过太多Python后端的接口代码了——Flask写的一套路由、FastAPI写的一套路由表面上都能跑通但打开路由表一看/api/get_user_list、/api/user/update、/api/create_order全来了。功能一点没少但每个接口都在问你是GET还是POST参数放哪错误怎么返回全靠文档口口相传。RESTful API设计这件事难的不是写代码而是把接口当契约来设计。今天这篇我把这几年做Python后端实践下来最核心的一套RESTful设计规范完整梳理一遍从资源建模、URL风格、HTTP方法语义、状态码、错误体、版本控制到分页筛选再到认证限流幂等和工程落地全部配合Python代码讲清楚。适合正在用Flask/FastAPI/Django写接口的人无论你是刚入门的小白还是被接口维护折磨过的老开发都能从中找到可以直接抄走的东西。1. 先纠正一下RESTful API最常见的三个理解偏差很多同学写RESTful接口靠的是感觉路径短一点、用上HTTP方法、返回JSON就觉得是RESTful了。但真到了联调阶段问题一个接一个冒出来。这背后往往是几个基础概念没捋清。1.1 第一个偏差REST约的是资源不是URLRESTful API的核心是资源Resource和资源的表述Representation。URL只是定位资源的一种手段但很多人的设计思路是URL好看 方法对了 REST结果路径全按操作来命名。举个例子用户上传头像这个功能。最容易写出来的方案是# FastAPI 错误示范 app.post(/api/upload_avatar) async def upload_avatar(user_id: int, file: UploadFile): ...看着没毛病但这不是资源式思维。上传头像本质上是更新用户这个资源里的头像字段更符合资源模型的做法是# 更好的建模头像作为用户资源的子资源 app.put(/users/{user_id}/avatar) async def update_avatar(user_id: int, file: UploadFile): ...第二个做法看着只是路径换了个写法但背后是两种完全不同的思考方式。前者是在设计功能清单后者是在设计资源集合。RESTful的资源建模要求你先问自己在这个系统里哪些东西是需要被独立访问、修改、删除的实体用户、订单、商品、评论……这些是资源。上传头像、发送邮件、触发任务这类动作要么是资源的子资源要么是某个资源的更新操作能不用动词就不用动词。用Python写接口时我习惯先画一张资源清单再开始写路由。列的时候按核心实体 从属实体分两层就够了超过两层嵌套直接拆出去改成查询参数。1.2 第二个偏差REST不等于CRUD动作请求是绕不开的见过不少团队把REST理解成对数据库表的增删改查然后所有操作都硬套HTTP方法。数据库里有个订单状态字段要取消订单怎么办DELETE /orders/{id}那订单数据整个没了审计记录也没了肯定不行。RESTful风格里对这类状态变更动作成年人的做法是能归为资源更新的用PATCH/PUT更新状态实在没法归类的创建一个子资源来承载这个动作。# 取消订单本质是更新订单状态 app.patch(/orders/{order_id}) async def update_order(order_id: int, payload: OrderUpdatePayload): ... # 但如果取消订单后面跟着一堆复杂的审核流程可以建模为动作子资源 app.post(/orders/{order_id}/cancellation) async def cancel_order(order_id: int): ...前者适合简单状态翻转后者适合背后有一堆业务流程要跑的场景。两条路都对关键看你有没有把为什么这么选想清楚。我见过一个项目把所有开箱、关箱、打印、重发邮件全部做成了/api/v1/orders/{id}/xxx的POST整个路由表看下来像一个RPC服务那就不用硬说自己是RESTful了。RESTful设计讲究能则资源化不能则显式动作化而不是所有东西都生搬硬套。1.3 第三个偏差无状态指的是服务器不存上下文不代表不需要认证另一个高频误区是RESTful要求无状态所以我们就不做登录态了。这句话听得我直摇头。REST的无状态Stateless约束的是服务器不能在内存里偷偷记一份某个客户端正在干什么的会话状态每个请求都得自包含。但认证信息恰恰是需要每个请求都带上的东西——Authorization: Bearer token就是最标准的无状态认证做法。# FastAPI 认证依赖每个受保护接口都显式依赖当前用户 from fastapi import Depends, HTTPException from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials security HTTPBearer() async def get_current_user(credentials: HTTPAuthorizationCredentials Depends(security)): token credentials.credentials user await decode_jwt_token(token) # 解析JWT不查服务端session if not user: raise HTTPException(status_code401, detail无效认证信息) return user app.get(/users/me) async def read_me(user Depends(get_current_user)): return userJWT自包含用户信息服务器不做状态存储每次请求带着它这完全符合无状态约束。搞清楚这三个偏差之后后面的设计规范才能落到实处。2. 资源建模与URL设计名词、复数、嵌套深度都要有章法URL是API给调用方的第一印象也是最容易暴露设计功力深浅的地方。我的经验是先有资源模型图再有URL。2.1 集合资源和单例资源复数主路径加ID标准写法是集合用复数名词单例用集合路径加ID。比如用户资源。用FastAPI展示完整的一套写法from fastapi import FastAPI, HTTPException, status from pydantic import BaseModel app FastAPI() class User(BaseModel): id: int name: str email: str app.get(/users, response_modellist[User]) async def list_users(skip: int 0, limit: int 20): # 返回用户集合 ... app.post(/users, status_codestatus.HTTP_201_CREATED) async def create_user(user: User): # 创建新用户 ... app.get(/users/{user_id}, response_modelUser) async def get_user(user_id: int): # 返回单例 ... app.patch(/users/{user_id}, response_modelUser) async def update_user(user_id: int, user: User): # 部分更新 ... app.delete(/users/{user_id}, status_codestatus.HTTP_204_NO_CONTENT) async def delete_user(user_id: int): ...这组路由是RESTful的根几乎任何系统都能对号入座。但我想强调一点路径参数别只用int。订单号、商品编号可能是字符串提前用Python的类型注解约束好避免运行期才报错。2.2 嵌套资源别超过两层层级就是归属关系用户的订单标准的嵌套写法是/users/{user_id}/orders语义是属于这个用户的订单集合。但很多项目一嵌套就上头写出/users/{user_id}/orders/{order_id}/items/{item_id}这种三层四层的路径。嵌套的每一层都代表一次归属链查询层级越深URL越脆。我自己的经验准则是最多嵌套两级超过两级就把后面的资源拆出去用查询参数表达归属。比如订单项可以直接做成/order-items?order_idxxx或者保持/orders/{order_id}/items但不要再往下挂了。选择嵌套还是扁平核心看归属是否天然且强约束。订单从属于用户这是天然归属嵌套没问题但用户给订单写评价这种评价的宿主其实是订单而不是用户我会写成/orders/{order_id}/reviews而不是/users/{user_id}/orders/{order_id}/review。2.3 命名风格全小写、连字符、不用动词、避免实现细节命名这块我遇到过太多次因不统一引发的低级问题。规范性建议直接记URL路径一律小写单词间用-连字符/user-addresses而非/user_addresses或/userAddresses。路径里只出现资源名词不出现动词动作要么交给HTTP方法要么做动作子资源。不出现/api/get_users.php这种带实现细节的写法。主键ID不在URL里暴露数据库自增ID线上环境尽量用UUID或不可枚举的业务编号。# 错误示范 app.get(/api/v1/getUserInfo) async def get_user_info(uid: int): ... # 正确示范 app.get(/v1/users/{user_id}, response_modelUser) async def get_user(user_id: str): ...UUID做用户ID会让URL变长但换来的是资源不可枚举防止别人遍历URL直接扒走全部用户数据。利弊权衡下来在高安全敏感的场景里UUID的收益远大于URL变长的代价。3. HTTP方法语义和状态码客户端只认协议不认你body里的那句话这是整个RESTful设计里信息量最大、也最容易被敷衍的部分。HTTP方法有各自的语义状态码有各自的含义。客户端是拿着协议标准来对接你的你把错误放进200的body里客户端根本不知道怎么统一处理。3.1 五个核心方法的正确打开方式方法语义幂等性典型用途Python响应GET查询资源幂等获取集合或单例200 OKPOST创建资源/触发动作非幂等创建新资源201 CreatedPUT整体替换资源幂等全量更新200 或 204PATCH部分更新资源非严格幂等更新部分字段200 或 204DELETE删除资源幂等删除204 No Content最经典的错误是更新用户资料为什么用POST。如果你更新一个用户的name字段POST发一遍再发一遍第二次请求服务端会因为你重复执行而产生不同的结果比如重复写日志、重复触发消息PUT则不同同一个请求发多少次最终资源状态都一样。这是幂等性的价值。PUT和PATCH的差别也常被忽略。PUT是把资源替换成我给你的这个完整版本客户端需要提交完整对象PATCH是只改我传给你的这几个字段不需要带上整份资源。# PUT全量替换客户端必须给全字段 app.put(/users/{user_id}) async def replace_user(user_id: str, payload: UserReplacePayload): ... # PATCH部分更新客户端只给要改的字段 app.patch(/users/{user_id}) async def patch_user(user_id: str, payload: UserPatchPayload): ...这条设计规范跟数据库操作容易混淆别把PUT当成UPDATE、把PATCH当成部分UPDATE。PUT是替换语义在大多数ORM场景下实现起来往往是先删后建或者整体覆写PATCH才是真正意义上的增量更新。用错的话PUT一个{name:x}结果把用户email给清空了这种事故我见过不止一次。3.2 状态码别乱用一张表说清楚常用选择状态码是协议的一部分不要自己发明。常用的就那十几个每个都要能说出理由。状态码含义什么时候用错误示范200请求成功GET/PATCH/PUT成功返回数据创建资源也返回200应该201201资源创建成功POST创建完成配合Location头更新操作返回201204请求成功无返回体DELETE成功、PUT成功DELETE返回200加空json301/302重定向资源迁移内部跳转也用302304缓存未修改配合ETag做条件请求业务失败返回304400客户端请求语法错误JSON格式错误、缺必填字段、参数类型错所有错误都返回400401未认证没带token、token失效权限不足时用401应用403403已认证但无权限登录用户访问无权资源一律返回无权限给客户端404资源不存在URL对不上或资源不存在拿404掩盖认证失败不推荐405方法不允许资源存在但没用对该方法路由没写就丢出500409资源状态冲突删除有子资源的订单、重复创建唯一约束什么都丢400422服务端能解析但语义错误字段格式对但业务规则校验失败如邮箱已被占用和400混用429请求过频触发限流返回500表示过载500服务端内部错误未捕获异常客户端参数错也丢500503服务不可用依赖服务挂掉、维护中正常业务报错用503实际用得最多的组合GET 200、POST 201、DELETE 204、参数问题400、资源不存在404、业务冲突409/422、认证失败401、权限不足403。这让客户端可以用极简的状态码判断逻辑处理绝大多数情况。3.3 422与400的正确分界线400和422的边界是我在代码评审里提到最多的问题。我的区分标准很明确400请求本身无法被解析比如JSON解析失败、类型错误、必填字段缺失——请先修请求。422请求能被解析但字段里的值不满足业务规则比如年龄填了-1、邮箱格式非法、唯一字段冲突——请求结构没问题内容是错的。from fastapi import FastAPI, Request, status from fastapi.responses import JSONResponse app.exception_handler(ValueError) async def value_error_handler(request: Request, exc: ValueError): return JSONResponse( status_codestatus.HTTP_422_UNPROCESSABLE_ENTITY, content{error: {code: UNPROCESSABLE_ENTITY, message: str(exc)}} )如果分不清宁可多做几个422也别全堆在400里。客户端遇到400会觉得自己请求格式有毛病但明明是传了一个重复的用户名格式没问题这对排查方向的影响很大。4. 版本、分页、过滤、排序这些细节决定接口能活多久前面那些是打地基这一节讲的是接口上线之后怎么活着、怎么演化。4.1 版本策略URL路径版本是最省心的方案版本这个问题不提前规划半年后就会遇到改了字段老客户端全挂的惨剧。行业主流有两种URL路径版本/v1/users、/v2/users。直观、跨网络设备兼容、没有任何协商成本绝大多数团队的首选。请求Header版本Accept: application/vnd.yourapp.v2json。更纯REST但调试麻烦网关过滤也不方便小团队用的人少。我的建议简单直接用URL路径版本从第一个接口就开始带上/v1。虽然看着多写几个字母但后续做破坏性升级时不用改逻辑直接加一个/v2就行。from fastapi import APIRouter v1_router APIRouter(prefix/v1) v2_router APIRouter(prefix/v2) v1_router.get(/users) async def list_users_v1(): return [{id: 1, name: old}] v2_router.get(/users) async def list_users_v2(): return [{id: 1, name: new, avatar_url: ...}] app.include_router(v1_router) app.include_router(v2_router)旧版本什么时候下线我一般定个规则至少保留两个大版本/v1和/v2共存/v1只修bug不打磨新功能通知周期过了再下掉。没有这个规则代码库里会堆出一堆没人维护的老接口最后谁都不敢删。4.2 分页page/limit适合管理端cursor适合C端分页这块两种主流方案各有适用场景。最常用的offset分页是page加limitGET /v1/orders?page1limit20。实现简单翻页跳页都方便而且能直接算出总页数。但它有两个先天缺陷数据量大时深翻页性能极差offset越大数据库需要扫描跳过的行越多新增数据插入时翻页会出现重复或跳过的数据。针对C端信息流一类的场景我更喜欢cursor分页GET /v1/orders?cursoreyJpZCI6MTAwfQlimit20。cursor通常是一段编码后的游标值代表从这个位置开始取下一页。class CursorPaginator: async def paginate(self, model, cursor: str | None, limit: int): query model.select() if cursor: # decode cursor 为上一页最后一条记录的 id last_id decode_cursor(cursor) query query.where(model.id last_id) rows await query.order_by(model.id.desc()).limit(limit).execute() next_cursor encode_cursor(rows[-1].id) if len(rows) limit else None return {items: rows, next_cursor: next_cursor}这种分页永远只往后翻不会因为新增数据导致重复而且性能稳定。缺点是跳页麻烦。实践里我会在管理后台用page分页运营要能直接跳到第50页在App端列表用cursor分页。4.3 过滤、排序、字段裁剪统一的query参数约定过滤和排序写得好能省掉一堆加个筛选条件就新写一个接口的迭代。过滤用相同的参数名运算符表达。GET /v1/orders?statuspaidamount_gte100created_at_lte2024-01-01。_lte、_gte、_ne这些后缀是约定俗成一眼能看懂。排序sortcreated_at升序sort-created_at降序多个字段用逗号分隔sort-created_at,name。字段裁剪fieldsid,name,email返回结果只包含这几个字段。这招对移动端省流量非常有用5G时代也越来越受欢迎。app.get(/v1/orders) async def list_orders( status: str | None None, amount_gte: float | None None, sort: str -created_at, fields: str id,total_amount,status ): query Order.select() if status: query query.where(Order.status status) if amount_gte is not None: query query.where(Order.total_amount amount_gte) # 解析 sort 参数构造 OrderBy # 解析 fields 参数控制响应字段 ...参数一多交给函数逐一接收会越写越长。FastAPI支持把这类查询参数收敛到Pydantic模型里统一处理。实际项目中我还会在API入口校验sort白名单防止客户端传任意字段排序拖垮数据库。5. 错误体和统一返回结构让调用方少写一堆if状态码解决了大类判断但真正让联调效率天差地别的是错误返回结构的统一程度。我见过最糟糕的API是有的错误返回{message: 新增成功}配个500你没看错有的返回{code: 1, msg: 系统繁忙}还有的干脆直接一个null。客户端对接三个服务得写三套解析逻辑。5.1 错误体长什么样才够用我建议统一采用下面这个结构基本思想参考RFC 7807但做了简化{ error: { code: ORDER_NOT_FOUND, message: 订单不存在或已被删除, details: [ {field: order_id, reason: 该订单ID无法在系统中找到} ], trace_id: a1b2c3d4 } }字段说明code机器可读的业务错误码客户端代码里分支判断就靠它别让前端去匹配中文message。message给人看的错误描述要具体不要写系统错误这种废话。details字段级错误列表专门用来承载422校验失败时每个字段的具体原因前端能直接定位到表单控件上。trace_id请求追踪ID出现问题时让用户截图报错后端拿这个ID一查日志整个请求链路都在。from fastapi import FastAPI, Request from fastapi.responses import JSONResponse class ApiError(Exception): def __init__(self, code: str, message: str, status_code: int 400, details: list | None None): self.code code self.message message self.status_code status_code self.details details or [] app.exception_handler(ApiError) async def api_error_handler(request: Request, exc: ApiError): return JSONResponse( status_codeexc.status_code, content{ error: { code: exc.code, message: exc.message, details: exc.details, trace_id: request.headers.get(X-Trace-ID, ) } } )有了这个统一异常类业务代码里碰到不合法的情况直接raise ApiError(...)框架层统一兜底转成错误结构不用每个接口自己写一遍try/except。5.2 业务错误码和HTTP状态码怎么分工这块容易走极端。一种团队完全没有业务错误码HTTP状态码当一切另一种正是大量互金、传统系统转过来的团队所有错误都返回200 codemsgHTTP状态码永远只有200。两者都有问题。只靠HTTP状态码无法表达订单已发货不可取消这种细粒度业务状态全部用200 code又在毁灭性地浪费HTTP协议本身的状态语义。中间的做法是状态码负责大类认证、权限、资源存在性、冲突、参数格式。业务错误码负责明细在这个大类里具体发生了什么。# 使用示例 app.post(/v1/orders/{order_id}/cancellation) async def cancel_order(order_id: int): order await get_order(order_id) if not order: raise ApiError(ORDER_NOT_FOUND, 订单不存在, status_code404) if order.status shipped: raise ApiError(ORDER_ALREADY_SHIPPED, 订单已发货无法取消, status_code409) ...这样客户端处理逻辑可以分层先根据状态码决定是否走通用错误弹窗再根据业务错误码决定是否走特殊流程比如弹充值页。如果直接用403表示订单已发货过两天你还会发现403里混进来用户没权限看这个订单就彻底拆不开了。5.3 校验错误的details结构Pydantic在做数据校验时能自动生成字段级错误把错误转换成统一结构很方便。FastAPI的RequestValidationError可以定制下面这个写法让前端的表单校验展示变成一件简单的事from fastapi.exceptions import RequestValidationError app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): details [] for err in exc.errors(): loc ..join(str(x) for x in err.get(loc, [])) details.append({field: loc, reason: err.get(msg, 校验失败)}) return JSONResponse( status_code422, content{ error: { code: VALIDATION_ERROR, message: 请求参数校验失败, details: details, trace_id: request.headers.get(X-Trace-ID, ) } } )这个handler一挂所有接口的参数校验错误都会自动统一成这套格式前端只用写一次解析函数就够了不用为每个页面单独适配错误弹窗。6. 认证、限流、幂等和缓存语义上线之后真正决定API质量的细节设计规范聊完讲点线上实战更容易踩坑的部分。接口上线后被高频问到的问题基本集中在这四个方向。6.1 认证方案选型JWT是通用解别自己造Session轮子Python生态里做认证JWT已经是绝对主流。它的好处和前面说的一样服务器无状态、多端登录天然支持、密钥签发即可。在FastAPI里配合python-jose库实现JWT签发和校验from jose import jwt, JWTError from datetime import datetime, timedelta SECRET_KEY your-secret-key-change-me ALGORITHM HS256 def create_access_token(user_id: str, expires_minutes: int 30) - str: expire datetime.utcnow() timedelta(minutesexpires_minutes) payload {sub: user_id, exp: expire} return jwt.encode(payload, SECRET_KEY, algorithmALGORITHM) async def decode_jwt_token(token: str) - dict: try: return jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) except JWTError: raise HTTPException(status_code401, detail无效认证信息)几个实际经验总结Token放在Authorization: Bearer token头里不要放body也不建议放URL查询参数会进日志。过期时间别太长建议15分钟到2小时配合Refresh Token使用。涉及高权限操作改密、提现时别省二次校验要求用户重新输入密码或走验证码流程。6.2 限流别让429这个状态码变成传说没有限流的API就像不设路障的停车场早晚要出事。限流常见做法是令牌桶算法每次请求从桶里取一个令牌令牌按固定速率补充。缓存到Redis每个客户端维度单独计数。import redis.asyncio as redis import time class RateLimiter: def __init__(self, redis_client, rate: int, capacity: int): self.redis redis_client self.rate rate # 每秒补充令牌数 self.capacity capacity # 桶容量 async def allow(self, key: str) - bool: async with self.redis.pipeline() as pipe: now time.time() pipe.multi() pipe.zremrangebyscore(key, 0, now - 1) pipe.zadd(key, {str(now): now}) pipe.zcard(key) result await pipe.execute() return result[-1] self.capacity配合FastAPI中间件在响应头里带上X-RateLimit-Remaining触限时返回429并设置Retry-After头客户端就能做到配合退避。app.middleware(http) async def rate_limit_middleware(request: Request, call_next): client_id request.headers.get(X-Client-ID, request.client.host) if not await limiter.allow(frate:{client_id}): return JSONResponse( status_code429, content{error: {code: RATE_LIMITED, message: 请求过于频繁请稍后重试}} ) return await call_next(request)注意一点限流维度不要只按IP。同一公司出口IP共用一个公网IP的情况很常见只按IP限流会误伤一批人。实际项目中我一般是IP限流兜底 用户级限流精细控制组合。6.3 幂等性POST接口也能变成幂等的前面说POST不幂等这是协议默认行为。但很多场景下比如客户端网络超时重试同一个创建订单请求发了两遍结果出现两个一模一样的订单用户在后台看得一头问号。解决办法是幂等键。客户端在创建请求时生成一个唯一的Idempotency-Key放在请求头里服务端收到后先查这个key有没有处理过处理过就直接返回第一次的结果没处理过就执行并存储结果。app.post(/v1/orders) async def create_order( order: OrderCreatePayload, idempotency_key: str Header(..., aliasIdempotency-Key) ): cached await redis.get(fidem:{idempotency_key}) if cached: return JSONResponse(status_code201, contentjson.loads(cached)) order await create_order_record(order) await redis.set(fidem:{idempotency_key}, order.json(), ex86400) return orderTTL我习惯设24小时覆盖客户端单次操作的完整生命周期。超过24小时重试算新请求对业务来说这个边界是合理的。6.4 缓存语义ETag与304的正确用法最后聊GET接口的缓存优化。很多人一谈缓存就上Redis其实HTTP层自带的条件请求机制在API场景下往往是更干净的方案。给资源生成一个ETag资源内容的哈希客户端下次请求时带上If-None-Match如果资源没变服务端直接返回304省掉body传输。import hashlib from fastapi.responses import Response async def generate_etag(data: dict) - str: raw json.dumps(data, sort_keysTrue).encode() return hashlib.md5(raw).hexdigest() app.get(/v1/users/{user_id}) async def get_user(user_id: str, if_none_match: str | None Header(defaultNone)): user await get_user_from_db(user_id) body user.model_dump_json() etag await generate_etag(json.loads(body)) if if_none_match and if_none_match etag: return Response(status_code304) return Response(contentbody, headers{ETag: etag})这里最容易被忽略的是304只是省流量数据库查询还是跑了一遍。真正的性能提升还得靠数据库层或Redis层把查询结果缓存起来HTTP条件请求负责的是回程流量。两者结合才叫完整的缓存方案。我自己在实际项目里固定的做法是接口一上线就把认证、限流、统一错误处理、日志trace_id四件套接入这几个属于不补就会出事的基础设施分页、排序、版本规划在建表定义API文档时一起设计好别等联调了才临时加。RESTful设计没有银弹但把这套约束固化到代码里、文档里和评审清单里团队维护接口的体感会有质的变化。如果你刚开始重构老接口建议不要一把梭挑一个最核心的资源先按这套规范重做跑通之后再逐步铺开——效果比推倒重来稳得多。
返回列表