安装
pip install grpcio -i https://pypi.tuna.tsinghua.edu.cn/simple -- 安装gRPC
pip install protobuf -i https://pypi.tuna.tsinghua.edu.cn/simple --安装ProtoBuf 相关的 python 依赖库
pip install grpcio-tools -i https://pypi.tuna.tsinghua.edu.cn/simple --安装python grpc 的 protobuf 编译工具
生成
python -m grpc_tools.protoc -I ./protos --python_out=. --pyi_out=. --grpc_python_out=. ./protos/ticket.proto
--python_out --pyi_out --grpc_python_out 对应生成的ticket_pb2.py ticket_pb2.pyi ticket_pb2.pyi三个文件对应路径
服务端
简单创建一个服务端,证书生成参考基与python的GRPC SSL/TLS加密及Token鉴权 - Nolinked - 博客园 (cnblogs.com)
import grpc
from concurrent import futures
import logging
import ticket_pb2,ticket_pb2_grpc
class Ticket(ticket_pb2_grpc.TicketServicer):
# 重写原先proto文件方法
def QueryTicket(self, request, context):
return ticket_pb2.QueryReply(num=100)
def serve():
with open('server.key','rb') as f:
private_key = f.read()
with open('server.crt','rb') as f:
certificate_chain = f.read()
server_credentials = grpc.ssl_server_credentials(
((private_key, certificate_chain,),))
port = '50051'
# 启用ssl
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
ticket_pb2_grpc.add_TicketServicer_to_server(Ticket(), server)
server.add_secure_port('[::]:' + port,server_credentials)
server.start()
print("Server started, listening on " + port)
server.wait_for_termination()
if __name__ == '__main__':
logging.basicConfig()
serve()
客户端
创建一个客户端于服务端通信
import grpc
import ticket_pb2,ticket_pb2_grpc
from src.google.protobuf.struct_pb2 import Struct
_HOST = 'localhost'
_PORT = '50051'
def run():
with open('server.crt', 'rb') as f:
trusted_certs = f.read()
credentials = grpc.ssl_channel_credentials(root_certificates=trusted_certs)
options = (('grpc.ssl_target_name_override','WLYNC',),)
with grpc.secure_channel((_HOST+':'+_PORT), credentials, options = options) as channel:
stub = ticket_pb2_grpc.TicketStub(channel)
response:ticket_pb2.QueryReply = stub.QueryTicket(ticket_pb2.QueryRequest(id="1"))
print(f"票数: {response.num}")
if __name__ == '__main__':
run()