跳到主要内容

规范API响应内容

前言​

在设计API返回内容时，通常需要与前端约定好API返回响应体内容的格式。这样方便前端进行数据反序列化时相应的解析处理，也方便其它服务调用。不同公司有不同的响应内容规范要求，这里以常见的JSON响应体为例：

{
    "code": 200,
    "data": {
        "content": "this is /a/1"
    },
    "msg": "success"
}

code字段​

code状态码主要用于表示错误类型区间状态码。如果设计比较简单，可以直接使用HTTP的状态码。如果是一个大型系统，也可以设计一套自定义的状态码。比如：

from enum import Enum

class BizStatus(Enum):
    # custom status code
    OK = 200
    BadRequestA1 = 4001  # 请求参数异常-A情况
    BadRequestA2 = 4002  # 请求参数异常-B情况

message字段​

message 字段是对当前 code 状态码错误明细的补充说明。通常不同的code状态码会有不同的message描述信息。

data字段​

data 值通常代表返回的响应体内容。

示例代码​

以下代码定义了一个JSON响应类，api在返回的时候需要引用这个响应类。除此之外，还对404和一般异常做了统一处理，当出现这两类异常时，也会返回JSON结构的响应体。

from flask import Flask, request, jsonify, make_response
from http import HTTPStatus

API_KEY_SVCA = "flask_unify_response"

app = Flask(__name__)


class JsonResponse:
    """A class to represent a JSON response."""
    def __init__(
        self, code: HTTPStatus = HTTPStatus.OK, msg: str = "success", data=None
    ):
        self.code = code
        self.msg = msg
        self.data = data

    def to_dict(self):
        return {
            "code": self.code.value,
            "msg": self.msg,
            "data": self.data,
        }

    def to_json(self):
        return jsonify(self.to_dict())

    def response(self):
        response = make_response(self.to_json(), self.code.value)
        response.headers["Content-Type"] = "application/json"
        return response


@app.errorhandler(404)
def error_handler_not_found(error):
    req_method = request.method
    req_path = request.path
    return JsonResponse(
        code=HTTPStatus.NOT_FOUND,
        msg=f"{req_method} {req_path} Not Found",
    ).response()


@app.errorhandler(Exception)
def error_handler_generic(error):
    req_method = request.method
    req_path = request.path
    return JsonResponse(
        code=HTTPStatus.INTERNAL_SERVER_ERROR,
        msg=f"Internal Server Error. {req_method} {req_path}",
        data={"error": str(error)},
    ).response()


@app.get("/a/1")
def apitest_a1():
    return JsonResponse(
        code=HTTPStatus.OK, msg="success", data={"content": "this is /a/1"}
    ).response()


@app.get("/a/2")
def apitest_a2():
    raise Exception("exception in a/2")


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=8001)

客户端请求测试:

$ curl -s http://127.0.0.1:8001/a/1 | python3 -m json.tool
{
    "code": 200,
    "data": {
        "content": "this is /a/1"
    },
    "msg": "success"
}


$ curl -s http://127.0.0.1:8001/a/2 | python3 -m json.tool
{
    "code": 500,
    "data": {
        "error": "exception in a/2"
    },
    "msg": "Internal Server Error. GET /a/2"
}


$ curl -s http://127.0.0.1:8001/a/3 | python3 -m json.tool
{
    "code": 404,
    "data": null,
    "msg": "GET /a/3 Not Found"
}

示例2-直接继承Response类​

from flask import Flask, Response, request
from http import HTTPStatus
from typing import Optional
import json
import traceback

app = Flask(__name__)

class JsonResponse(Response):
    def __init__(self, data: Optional[dict] = None, code: HTTPStatus = HTTPStatus.OK, msg: str = "success"):
        headers = dict({"Content-Type": "application/json; charset=utf-8"})
        response = json.dumps({
            "code": code.value,
            "msg": msg,
            "data": data,
        })
        super().__init__(response=response, status=code.value, headers=headers)
    

class Success(JsonResponse):
    """http status code: 200"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"{request.method} {request.path} success"
        super().__init__(code=HTTPStatus.OK, msg=msg, data=data)

class Fail(JsonResponse):
    """http status code: 500"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Fail to {request.method} {request.path}"
        super().__init__(code=HTTPStatus.INTERNAL_SERVER_ERROR, msg=msg, data=data)

class ArgumentNotFound(JsonResponse):
    """http status code: 400"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Argument not found when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.BAD_REQUEST, msg=msg, data=data)

class ArgumentInvalid(JsonResponse):
    """http status code: 400"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Invalid arguments when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.BAD_REQUEST, msg=msg, data=data)

class AuthFailed(JsonResponse):
    """http status code: 401"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Auth failed when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.UNAUTHORIZED, msg=msg, data=data)

class ResourceConflict(JsonResponse):
    """http status code: 409"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Resource conflict when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.CONFLICT, msg=msg, data=data)

class ResourceNotFound(JsonResponse):
    """http status code: 404"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"{request.method} {request.path} not found"
        super().__init__(code=HTTPStatus.NOT_FOUND, msg=msg, data=data)

class ResourceForbidden(JsonResponse):
    """http status code: 403"""
    def __init__(self, data: Optional[dict] = None):
        msg = f"Resource forbidden when {request.method} {request.path}"
        super().__init__(code=HTTPStatus.FORBIDDEN, msg=msg, data=data)

def error_handler_notfound(error):
    return ResourceNotFound()

def error_handler_generic(error):
    print(str(traceback.format_exc()))
    return Fail()

@app.get("/")
def index():
    return Success("helloworld")

@app.get("/b")
def index_b():
    return Fail()

@app.get("/c")
def index_c():
    return ArgumentNotFound()

@app.get("/d")
def index_d():
    return ArgumentInvalid()

@app.get("/e")
def index_e():
    return AuthFailed()

@app.get("/f")
def index_f():
    return ResourceConflict()

@app.get("/g")
def index_g():
    return ResourceNotFound()



if __name__ == '__main__':
    app.errorhandler(Exception)(error_handler_generic)
    app.errorhandler(404)(error_handler_notfound)
    app.run(host="127.0.0.1", port=8000, debug=False)

处理datetime类型问题​

问题描述：datetime类型不能直接序列化成json，需要改下json编码器

from flask import Flask, Response
from http import HTTPStatus
from typing import Optional
import json

from datetime import datetime

app = Flask(__name__)

class CustomJSONEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, datetime):
            return o.isoformat()
        return super().default(o)

class JsonResponse(Response):
    def __init__(self, data: Optional[dict] = None, code: HTTPStatus = HTTPStatus.OK, msg: str = "success"):
        headers = dict({"Content-Type": "application/json; charset=utf-8"})
        response = json.dumps({
            "code": code.value,
            "msg": msg,
            "data": data,
        },cls=CustomJSONEncoder)
        super().__init__(response=response, status=code.value, headers=headers)

@app.get("/")
def index():
    data = {
        "now": datetime.now(),
        "sth": "hahahaha",
    }
    return JsonResponse(data=data)

if __name__ == "__main__":
    app.run(host="127.0.0.1", port=5000, debug=False)