基于含Well-Known Timestamp的.proto生成Python gRPC客户端的技术咨询

我来一步步帮你搞定这个问题，你之前的proto文件只差几个关键调整就能完美集成Timestamp了！

第一步：修正你的.proto文件 #

你之前注释掉的导入语句路径不对，Well-Known类型的正确导入路径是google/protobuf/timestamp.proto，同时要把字段声明也恢复。修改后的hello.proto应该是这样：

syntax = "proto3";
package helloworld;

// 正确导入Well-Known Timestamp类型
import "google/protobuf/timestamp.proto";

service Greeter {
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

message HelloRequest {
  string name = 1;
  // 添加Timestamp字段
  google.protobuf.Timestamp smth = 2;
}

message HelloReply {
  string message = 1;
  // 也可以在响应里添加Timestamp字段（可选）
  google.protobuf.Timestamp response_time = 2;
}

第二步：重新生成gRPC代码 #

直接用你之前的命令重新生成pb2和grpc文件就行——不用额外下载任何proto文件，grpc_tools已经内置了所有Well-Known类型的定义，它会自动找到google/protobuf/timestamp.proto：

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./hello.proto

第三步：在Python代码中使用Timestamp #

生成代码后，你就可以在客户端和服务端里轻松使用Timestamp了：

客户端构造带Timestamp的请求 #

首先导入Timestamp类，然后有几种常见的创建方式：

import hello_pb2
import hello_pb2_grpc
from google.protobuf.timestamp_pb2 import Timestamp
from datetime import datetime
import grpc

def run():
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = hello_pb2_grpc.GreeterStub(channel)
        # 创建请求对象
        request = hello_pb2.HelloRequest(name="Bohdan")
        
        # 方式1：获取当前时间（UTC）
        request.smth.GetCurrentTime()
        
        # 方式2：从datetime对象转换（推荐用UTC时间避免时区问题）
        # request.smth.FromDatetime(datetime.utcnow())
        
        # 方式3：手动设置秒和纳秒
        # request.smth.seconds = 1718000000
        # request.smth.nanos = 300000000
        
        # 发送请求
        response = stub.SayHello(request)
        print(f"Received: {response.message}")
        # 如果响应里也有Timestamp，可以转换为datetime查看
        if response.HasField("response_time"):
            print(f"Response time: {response.response_time.ToDatetime()}")

if __name__ == '__main__':
    run()

服务端处理Timestamp字段 #

服务端读取Timestamp后，可以转换成Python的datetime对象方便处理：

import hello_pb2
import hello_pb2_grpc
from google.protobuf.timestamp_pb2 import Timestamp
from datetime import datetime
import grpc
from concurrent import futures

class Greeter(hello_pb2_grpc.GreeterServicer):
    def SayHello(self, request, context):
        # 读取请求中的Timestamp并转换为datetime
        received_time = request.smth.ToDatetime()
        print(f"Received request from {request.name} at {received_time}")
        
        # 构造响应，也可以添加Timestamp
        response = hello_pb2.HelloReply(message=f"Hello, {request.name}!")
        response.response_time.GetCurrentTime()
        return response

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    hello_pb2_grpc.add_GreeterServicer_to_server(Greeter(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    server.wait_for_termination()

if __name__ == '__main__':
    serve()

注意事项 #

• Timestamp默认使用UTC时间，所以尽量用datetime.utcnow()而不是datetime.now()来转换，避免时区错乱；
• 如果需要处理时区，可以先用第三方库（比如pytz）将本地时间转换为UTC，再传给Timestamp；
• 调用HasField()方法可以检查字段是否被设置，避免访问未初始化的Timestamp导致错误。

内容的提问来源于stack exchange，提问作者bohdan_trotsenko