python安装grpc相关库
1 | pip install grpc grpcio grpcio-tools |
python中引入第三方proto文件
下载第三方proto文件到你本地proto目录，最好把路径都改到一级目录，省去很多不必要的麻烦
生成.pb2.py和.pb2_grpc.py文件：
1 | python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. http.proto |
如果使用目标pb2文件过程中依赖第三方的pb2文件，可以再次执行上面 python -m 生成第三方的pb2文件。
python设置grpc请求头headers：
查阅了网上所有的资料都没有看到定制headers的方法，最终在grpc python example代码中发现了一种使用拦截器的方法，有点麻烦，不过能解决问题，如果你发现更方便的方法，麻烦告知我一下，万分感谢，拦截器方法如下：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75import grpc
import collections
class _GenericClientInterceptor(
grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor,
grpc.StreamUnaryClientInterceptor, grpc.StreamStreamClientInterceptor):
def __init__(self, interceptor_function):
self._fn = interceptor_function
def intercept_unary_unary(self, continuation, client_call_details, request):
new_details, new_request_iterator, postprocess = self._fn(
client_call_details, iter((request,)), False, False)
response = continuation(new_details, next(new_request_iterator))
return postprocess(response) if postprocess else response
def intercept_unary_stream(self, continuation, client_call_details,
request):
new_details, new_request_iterator, postprocess = self._fn(
client_call_details, iter((request,)), False, True)
response_it = continuation(new_details, next(new_request_iterator))
return postprocess(response_it) if postprocess else response_it
def intercept_stream_unary(self, continuation, client_call_details,
request_iterator):
new_details, new_request_iterator, postprocess = self._fn(
client_call_details, request_iterator, True, False)
response = continuation(new_details, new_request_iterator)
return postprocess(response) if postprocess else response
def intercept_stream_stream(self, continuation, client_call_details,
request_iterator):
new_details, new_request_iterator, postprocess = self._fn(
client_call_details, request_iterator, True, True)
response_it = continuation(new_details, new_request_iterator)
return postprocess(response_it) if postprocess else response_it
def create(intercept_call):
return _GenericClientInterceptor(intercept_call)
class _ClientCallDetails(
collections.namedtuple(
'_ClientCallDetails',
('method', 'timeout', 'metadata', 'credentials')),
grpc.ClientCallDetails):
pass
def header_adder_interceptor(header, value):
def intercept_call(client_call_details, request_iterator, request_streaming,
response_streaming):
metadata = []
if client_call_details.metadata is not None:
metadata = list(client_call_details.metadata)
metadata.append((
header,
value,
))
client_call_details = _ClientCallDetails(
client_call_details.method, client_call_details.timeout, metadata,
client_call_details.credentials)
return client_call_details, request_iterator, None
return create(intercept_call)
server_address='www.example.com:50000'
with open('server.pem', 'rb') as f:
creds = grpc.ssl_channel_credentials(f.read())
token = '123'
header_adder_interceptor = header_adder_interceptor('authorization', 'token {}'.format(token))
channel = grpc.secure_channel(server_address, creds) #使用https证书的方法，需要可以参考一下，不需要证书可以使用grpc.insecure_channel()
intercept_channel = grpc.intercept_channel(channel, header_adder_interceptor)
stub = resume_pb2_grpc.ResumeStub(intercept_channel)具体使用过程中可以把header_adder_interceptor抽象出来变成个服务
还有个请求地址的问题，记住https证书只针对域名，不会针对ip，所以地址必须写成域名，本地测试可以把域名到ip的映射写到hosts文件中