当前位置：首页 > news >正文

Python项目中Protocol Buffers的应用示例

news 2025/7/15 12:00:36

项目结构

文件介绍

generated目录用于存储编译后的pb2文件
proto/user.proto定义的user协议
client.py模拟前后端分离向后端发起请求
server.py利用Flask创建简易web应用提供获取数据接口

依赖安装

protobuf==4.25.3
flask==3.0.2
flask-cors==4.0.0

(base) PS ......\protobuf-demo> conda activate protobuf-demo
(protobuf-demo) ......\protobuf-demo> pip install -r .\requirements.txt

proto编译

(protobuf-demo) PS D:\workspace\projects-py\protobuf-demo> protoc -I=proto --python_out=generated proto/user.proto

执行代码后将会在generated目录中生成user.proto对应的pb2文件

文件拆解

proto/user.proto

syntax = "proto3";package user;message User {int32 id = 1;string name = 2;string email = 3;Role role = 4;enum Role {GUEST = 0;USER = 1;ADMIN = 2;}
}message UserRequest {int32 user_id = 1;}message UserResponse {User user = 1;}message UserListResponse {repeated User users = 1;}

user_pb2.py

import requestsfrom generated import user_pb2def get_user(user_id, sub_type='json'):print(f"Getting user {user_id} in {sub_type} format......")response = requests.get(url=f"http://localhost:5000/api/users/{user_id}",headers={'Accept': f'application/{sub_type}'})if response.status_code == 200:if sub_type == 'json':print(response.json())if sub_type == 'protobuf':# 反序列化user_response = user_pb2.UserResponse()user_response.ParseFromString(response.content)user = user_response.userprint(f"ID: {user.id}\n"f"Name: {user.name}\n"f"Email: {user.email}\n"f"Role: {user_pb2.User.Role.Name(user.role)}")else:print(f"Error: {response.status_code} - {response.text}")def list_users(sub_type='json'):print(f"Listing users in {sub_type} format......")response = requests.get(url=f"http://localhost:5000/api/users",headers={'Accept': f'application/{sub_type}'})if response.status_code == 200:if sub_type == 'json':print(response.json())if sub_type == 'protobuf':# 反序列化user_list_response = user_pb2.UserListResponse()user_list_response.ParseFromString(response.content)users = user_list_response.usersfor user in users:print(f"ID: {user.id}\n"f"Name: {user.name}\n"f"Email: {user.email}\n"f"Role: {user_pb2.User.Role.Name(user.role)}")else:print(f"Error: {response.status_code} - {response.text}")if __name__ == '__main__':get_user(1, 'json')get_user(1, 'protobuf')list_users('json')list_users('protobuf')

server.py

from flask import Flask, request, Response
from flask_cors import CORSfrom generated import user_pb2app = Flask(__name__)
CORS(app)# 模拟数据库
users_db = {1: user_pb2.User(id=1, name='张三', email='zhangsan@example.com', role=user_pb2.User.Role.ADMIN),2: user_pb2.User(id=2, name='李四', email='lisi@example.com', role=user_pb2.User.Role.USER),3: user_pb2.User(id=3, name='王五', email='wangwu@example.com', role=user_pb2.User.Role.GUEST),
}@app.route('/api/users/<int:user_id>', methods=['GET'])
def get_user(user_id):# 检查客户端接受的Content-typeaccept_header = request.headers.get('Accept', '')if user_id in users_db:user = users_db[user_id]# 确认返回数据格式if 'application/protobuf' in accept_header:# 返回protobuf格式response = user_pb2.UserResponse()response.user.CopyFrom(user)# 序列化返回return Response(response.SerializeToString(), content_type='application/protobuf')else:# 默认返回JSON格式return {'id': user.id,'name': user.name,'email': user.email,'role': user_pb2.User.Role.Name(user.role),}else:return {'error': 'User not found'}, 404@app.route('/api/users', methods=['GET'])
def list_users():# 检查客户端接受的Content-typeaccept_header = request.headers.get('Accept', '')if 'application/protobuf' in accept_header:# 返回protobuf格式response = user_pb2.UserListResponse()for user in users_db.values():response.users.append(user)# 序列化返回return Response(response.SerializeToString(), content_type='application/protobuf')else:# 默认返回JSON格式return [{'id': user.id,'name': user.name,'email': user.email,'role': user_pb2.User.Role.Name(user.role),}for user in users_db.values()]if __name__ == '__main__':app.run(debug=True, port=5000)

client.py

import requestsfrom generated import user_pb2def get_user(user_id, sub_type='json'):print(f"Getting user {user_id} in {sub_type} format......")response = requests.get(url=f"http://localhost:5000/api/users/{user_id}",headers={'Accept': f'application/{sub_type}'})if response.status_code == 200:if sub_type == 'json':print(response.json())if sub_type == 'protobuf':# 反序列化user_response = user_pb2.UserResponse()user_response.ParseFromString(response.content)user = user_response.userprint(f"ID: {user.id}\n"f"Name: {user.name}\n"f"Email: {user.email}\n"f"Role: {user_pb2.User.Role.Name(user.role)}")else:print(f"Error: {response.status_code} - {response.text}")def list_users(sub_type='json'):print(f"Listing users in {sub_type} format......")response = requests.get(url=f"http://localhost:5000/api/users",headers={'Accept': f'application/{sub_type}'})if response.status_code == 200:if sub_type == 'json':print(response.json())if sub_type == 'protobuf':# 反序列化user_list_response = user_pb2.UserListResponse()user_list_response.ParseFromString(response.content)users = user_list_response.usersfor user in users:print(f"ID: {user.id}\n"f"Name: {user.name}\n"f"Email: {user.email}\n"f"Role: {user_pb2.User.Role.Name(user.role)}")else:print(f"Error: {response.status_code} - {response.text}")if __name__ == '__main__':get_user(1, 'json')get_user(1, 'protobuf')list_users('json')list_users('protobuf')

环境启动

启动服务端

(protobuf-demo) PS D:\workspace\projects-py\protobuf-demo> python .\server.py

启动客户端

(protobuf-demo) PS D:\workspace\projects-py\protobuf-demo> python .\client.py

结果输出如下

Getting user 1 in json format......
{'email': 'zhangsan@example.com', 'id': 1, 'name': '张三', 'role': 'ADMIN'}
Getting user 1 in protobuf format......
ID: 1
Name: 张三
Email: zhangsan@example.com
Role: ADMIN
Listing users in json format......
[{'email': 'zhangsan@example.com', 'id': 1, 'name': '张三', 'role': 'ADMIN'}, {'email': 'lisi@example.com', 'id': 2, 'name': '李四', 'role': 'USER'}, {'email': 'wangwu@example.com', 'id': 3, 'name': '王五', 'role': 'GUEST'}]
Listing users in protobuf format......
ID: 1
Name: 张三
Email: zhangsan@example.com
Role: ADMIN
ID: 2
Name: 李四
Email: lisi@example.com
Role: USER
ID: 3
Name: 王五
Email: wangwu@example.com
Role: GUEST