关于统一响应格式的补充

引言

在设计接口的时候，在前端调用通常使用的是状态码，比如 200、400、500等，但是标准的 HTTP 状态码却无法准确的表达业务的问题。

比如，登录请求返回的状态码，同样是 401，需要表达用户名或密码错误、验证码错误等多种业务场景。

考虑参照各大厂商规范，制定新的接口响应规范。

状态码 VS 自定义 Code

以国内腾讯、阿里、京东、微博为首的 API 响应规范如下：

在《阿里巴巴开发手册(泰山版)》中已经给出了通用的业务 code 规范。

json

{
  "code": 0,
  "message": "OK",
  "data": {}
}

所有接口的状态码都是 200，通过 code 区分业务。在微信小程序里尤甚，wx.request 的响应，401、500 也算 success，只有无响应(数据包丢了)才走 error。

新规范

在综合参考了 Google、Microsoft、Twitter 的接口设计后，对比国内的 API 形式，跟随 Google 的设计规范，仍然沿用状态码的形式。(有人说 Microsoft 的规范更好，但是我没太看懂)

Google API设计规范：https://cloud.google.com/apis/design/errors?hl=zh-cn

请求成功(状态码为200时)的设计不变：

[{
  "id": 1,
  "name": "Hello Kitty"
}, {
  "id": 2,
  "name": "史努比"
}]

失败时添加 state 状态与 message 错误详细信息：

json

{
  "state": "INVALID_ARGUMENT",
  "message": "密码不能为空"
}

HTTP 状态码	状态	错误详细信息
400	`INVALID_ARGUMENT`	密码不能为空
401	`CODE_INVALID`	验证码无效
401	`CREDENTIALS_INVALID`	用户名或密码错误

代码修改

app/common/response.py

python

from enum import Enum
from typing import Optional, Any

from fastapi.responses import JSONResponse

from pydantic import BaseModel


class APIResponseModel(BaseModel):
    """ API 响应模型 """
    code: int
    msg: str
    data: Optional[Any]


class APIResponse(JSONResponse):
    """
    API 响应，继承自JSONResponse，请求状态码全部为200，通过code区分业务
    """
    def __init__(self, code: int, msg: str, data: Optional[Any] = None):
        super().__init__(status_code=200, content=APIResponseModel(code=code, msg=msg, data=data).dict(exclude_none=True))


class APICode(Enum):
    """ API 响应状态码  """
    RUNTIME_ERROR = {'code': 500, 'msg': 'Runtime Error'}
    SUCCESS = {'code': 200, 'msg': 'Success'}
    INVALID_PARAMS = {'code': 400, 'msg': 'Invalid Params'}
    USER_NOT_FOUND = {'code': 404, 'msg': 'User Not Found'}
    USERNAME_EXISTED = {'code': 200, 'msg': 'Username Existed'}

    @property
    def code(self) -> int:
        return self.value.get('code')

    @property
    def msg(self) -> str:
        return self.value.get('msg')

app/common/exception.py

python

class APIException(Exception):
    """ API 异常类 """

    def __init__(self, code, msg, **kwargs):
        self.code = code
        self.msg = msg
        self.kwargs = kwargs

    def __repr__(self) -> str:
        class_name = self.__class__.__name__
        return f"{class_name}(error_code={self.code!r}, error_msg={self.msg!r}, kwargs={self.kwargs!r})"


class RuntimeException(Exception):
    """ 运行时异常类 """

    def __init__(self, code, msg, **kwargs):
        self.code = code
        self.msg = msg
        self.kwargs = kwargs

    def __repr__(self) -> str:
        class_name = self.__class__.__name__
        return f"{class_name}(error_code={self.code!r}, error_msg={self.msg!r}, kwargs={self.kwargs!r})"

app/common/error_handle.py

python

from typing import Sequence
from pydantic import parse_obj_as

from sqlalchemy.ext.asyncio import AsyncSession

from app.common.exception import APIException
from app.common.response import APICode
from app.schemas.user import UserCreate, UserUpdate, UserOut
from app.repository.user import UserRepository


class UserService:
    """ 用户服务类 """

    def __init__(self, db_session: AsyncSession):
        self.user_repository = UserRepository(db_session)

    async def create_user(self, user: UserCreate) -> UserOut:
        db_user = await self.user_repository.query_user_by_username(user.username)
        if db_user is not None:
            raise APIException(APICode.USERNAME_EXISTED.code, APICode.USERNAME_EXISTED.msg)
        return parse_obj_as(UserOut, await self.user_repository.create_user(user))

    async def query_user(self, user_id: int) -> UserOut:
        db_user = await self.user_repository.query_user(user_id)
        if db_user is None:
            raise APIException(APICode.USER_NOT_FOUND.code, APICode.USER_NOT_FOUND.msg)
        return parse_obj_as(UserOut, db_user)

    async def query_user_by_username(self, username: str) -> UserOut:
        db_user = await self.user_repository.query_user_by_username(username)
        if db_user is None:
            raise APIException(APICode.USER_NOT_FOUND.code, APICode.USER_NOT_FOUND.msg)
        return parse_obj_as(UserOut, db_user)

    async def query_users(self, skip: int = 0, limit: int = 100) -> Sequence[UserOut]:
        return parse_obj_as(Sequence[UserOut], await self.user_repository.query_users(skip, limit))

    async def update_user(self, user_id: int, user: UserUpdate) -> UserOut:
        cur_user = await self.user_repository.query_user(user_id)
        if cur_user is None:
            raise APIException(APICode.USER_NOT_FOUND.code, APICode.USER_NOT_FOUND.msg)
        return parse_obj_as(UserOut, await self.user_repository.update_user(user_id, user))

    async def delete_user(self, user_id: int) -> None:
        return await self.user_repository.delete_user(user_id)

app/service/user.py

python

from typing import Sequence
from pydantic import parse_obj_as

from sqlalchemy.ext.asyncio import AsyncSession

from app.common.exception import APIException
from app.common.response import APICode
from app.schemas.user import UserCreate, UserUpdate, UserOut
from app.repository.user import UserRepository


class UserService:
    """ 用户服务类 """

    def __init__(self, db_session: AsyncSession):
        self.user_repository = UserRepository(db_session)

    async def create_user(self, user: UserCreate) -> UserOut:
        db_user = await self.user_repository.query_user_by_username(user.username)
        if db_user is not None:
            raise APIException(APICode.USERNAME_EXISTED.code, APICode.USERNAME_EXISTED.msg)
        return parse_obj_as(UserOut, await self.user_repository.create_user(user))

    async def query_user(self, user_id: int) -> UserOut:
        db_user = await self.user_repository.query_user(user_id)
        if db_user is None:
            raise APIException(APICode.USER_NOT_FOUND.code, APICode.USER_NOT_FOUND.msg)
        return parse_obj_as(UserOut, db_user)

    async def query_user_by_username(self, username: str) -> UserOut:
        db_user = await self.user_repository.query_user_by_username(username)
        if db_user is None:
            raise APIException(APICode.USER_NOT_FOUND.code, APICode.USER_NOT_FOUND.msg)
        return parse_obj_as(UserOut, db_user)

    async def query_users(self, skip: int = 0, limit: int = 100) -> Sequence[UserOut]:
        return parse_obj_as(Sequence[UserOut], await self.user_repository.query_users(skip, limit))

    async def update_user(self, user_id: int, user: UserUpdate) -> UserOut:
        cur_user = await self.user_repository.query_user(user_id)
        if cur_user is None:
            raise APIException(APICode.USER_NOT_FOUND.code, APICode.USER_NOT_FOUND.msg)
        return parse_obj_as(UserOut, await self.user_repository.update_user(user_id, user))

    async def delete_user(self, user_id: int) -> None:
        return await self.user_repository.delete_user(user_id)

app/controller/user.py

python

from fastapi import APIRouter, Depends

from app.schemas.user import UserCreate, UserUpdate
from app.core.depends.service import get_user_service
from app.service.user import UserService
from app.common.response import APIResponse, APICode, APIResponseModel


class UserController:
    router = APIRouter()

    @staticmethod
    @router.post('/users/', response_model=APIResponseModel)
    async def create_user(user: UserCreate, user_service: UserService = Depends(get_user_service)):
        user = await user_service.create_user(user)
        return APIResponse(code=APICode.SUCCESS.code, msg=APICode.SUCCESS.msg, data=user)

    @staticmethod
    @router.get('/users/id/{user_id}', response_model=APIResponseModel)
    async def get_user(user_id: int, user_service: UserService = Depends(get_user_service)):
        user = await user_service.query_user(user_id)
        return APIResponse(code=APICode.SUCCESS.code, msg=APICode.SUCCESS.msg, data=user)

    @staticmethod
    @router.get('/users/username/{username}', response_model=APIResponseModel)
    async def get_user_by_username(username: str, user_service: UserService = Depends(get_user_service)):
        user = await user_service.query_user_by_username(username)
        return APIResponse(code=APICode.SUCCESS.code, msg=APICode.SUCCESS.msg, data=user)

    @staticmethod
    @router.get('/users', response_model=APIResponseModel)
    async def get_users(skip: int = 0, limit: int = 100, user_service: UserService = Depends(get_user_service)):
        users = await user_service.query_users(skip, limit)
        return APIResponse(code=APICode.SUCCESS.code, msg=APICode.SUCCESS.msg, data=users)

    @staticmethod
    @router.put('/users/{user_id}', response_model=APIResponseModel)
    async def update_user(user_id: int, user: UserUpdate, user_service: UserService = Depends(get_user_service)):
        user = await user_service.update_user(user_id, user)
        return APIResponse(code=APICode.SUCCESS.code, msg=APICode.SUCCESS.msg, data=user)

    @staticmethod
    @router.delete('/users/{user_id}', response_model=APIResponseModel)
    async def delete_user(user_id: int, user_service: UserService = Depends(get_user_service)):
        await user_service.delete_user(user_id)
        return APIResponse(code=APICode.SUCCESS.code, msg=APICode.SUCCESS.msg)

    @classmethod
    def get_router(cls):
        return cls.router

关于统一响应格式的补充 #

引言 #

状态码 VS 自定义 Code #

新规范 #

代码修改 #

关于统一响应格式的补充

引言

状态码 VS 自定义 Code

新规范

代码修改