Flask protobuf

Protobuf

基本介绍

Protobuf(Google Protocol Buffers)是google开发的的一套用于数据存储，网络通信时用于协议编解码的工具库.它和XML和Json数据差不多,把数据以某种形式保存起来.Protobuf相对与XML和Json的不同之处，它是一种二进制的数据格式，具有更高的传输，打包和解包效率。Protobuf 提供了C++、java、python、nodejs、php、go语言的支持，提供了windows(proto.exe)和linux平台动态编译生成proto文件对应的源文件。proto文件定义了协议数据中的实体结构(message ,field)。因为有google背书，所以在各个语言中被迅速普及，常见的应用场景就是微服务架构和DDD架构！！！

protoBuf下载地址：https://github.com/protocolbuffers/protobuf/releases

flask中基于protobuf实现api接口

环境搭建

# 安装格式转换工具，不同的操作系统，转换工具不同。windows->proto.exe，Linux->protobuf-compiler
sudo apt install -y protobuf-compiler
pip install flask
# python下面用于对protobuf进行编码解码的模块
pip install protobuf
# 网络请求工具库
pip install requests

接口代码

编写Protobuf数据数据，Protobuf_Message.proto，代码：

syntax = "proto3";

// 响应体
message Response {
  int32 returncode = 1;
  message data {
    string username = 1;
    string email = 2;
    int32 age = 3;
  }
  repeated data dataList = 2;
  string message = 3;
}

// 请求体
message Request {
  string data = 1;
  int32 page = 2;
  int32 pageSize = 3;
}

编译转换成python格式

protoc --python_out=./ Protobuf_Message.proto

服务端接口，server.py，代码：

from Protobuf_Message_pb2 import Request, Response
from flask import Flask, request
app=Flask(__name__)

@app.route("/api/demo", methods=["POST"])
def demo():
    # 解析请求
    request_data = Request()
    request_data.ParseFromString(request.get_data())

    print("data",request_data.data) # data user
    print("page",request_data.page) # page 1
    print("pageSize",request_data.pageSize) # pageSize 10

    # 编写响应信息
    response = Response()
    response.returncode = 200
    response.message = "Success"

    data_list = [
        {"username":"小明", "age": 18, "email":"xiaoming@mofang.com"},
        {"username":"小白", "age": 18, "email":"xiaobai@mofang.com"},
    ]

    for data in data_list:
        item = response.data()
        item.username=data["username"]
        item.age=data["age"]
        item.email=data["email"]
        response.dataList.append(item)

    return response.SerializeToString(), 200

if __name__ == '__main__':
    app.run("127.0.0.1", port=5000)

客户端，client.py，代码：

from Protobuf_Message_pb2 import Request, Response
import requests
def test_protobuf_api():
    """
    测试 接口
    :return:
    """
    request_data = Request()
    # 编写请求信息
    request_data.data = "user"
    request_data.page = 1
    request_data.pageSize = 10
    req_data = request_data.SerializeToString()# 序列化
    response = requests.post("http://127.0.0.1:5000/api/demo", data=req_data)

    # 获取响应内容
    res = Response()
    res.ParseFromString(response.content) # 反序列化
    print(res.returncode) # 200
    print(res.message)    # 成功
    for data in res.dataList:
        print(f"username:{data.username}, age:{data.age}, email: {data.email}")
        # username:小明, age:18, email=xiaoming@mofang.com
        # username:小白, age:18, email=xiaobai@mofang.com
if __name__ == '__main__':
    test_protobuf_api()

先有运行服务端代码，再运行客户端代码，即可查看效果。