The C based gRPC (C++, Python, Ruby, Objective-C, PHP, C#) https://grpc.io/
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

171 lines
6.8 KiB

#!/usr/bin/env python
# Copyright 2016 gRPC authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import pkg_resources
import sys
import os
from grpc_tools import _protoc_compiler
_PROTO_MODULE_SUFFIX = "_pb2"
_SERVICE_MODULE_SUFFIX = "_pb2_grpc"
_DISABLE_DYNAMIC_STUBS = "GRPC_PYTHON_DISABLE_DYNAMIC_STUBS"
def main(command_arguments):
"""Run the protocol buffer compiler with the given command-line arguments.
Args:
command_arguments: a list of strings representing command line arguments to
`protoc`.
"""
command_arguments = [argument.encode() for argument in command_arguments]
return _protoc_compiler.run_main(command_arguments)
# NOTE(rbellevi): importlib.abc is not supported on 3.4.
if sys.version_info >= (3, 5, 0):
import contextlib
import importlib
import importlib.machinery
import threading
_FINDERS_INSTALLED = False
_FINDERS_INSTALLED_LOCK = threading.Lock()
def _maybe_install_proto_finders():
global _FINDERS_INSTALLED
with _FINDERS_INSTALLED_LOCK:
if not _FINDERS_INSTALLED:
sys.meta_path.extend([
ProtoFinder(_PROTO_MODULE_SUFFIX,
_protoc_compiler.get_protos),
ProtoFinder(_SERVICE_MODULE_SUFFIX,
_protoc_compiler.get_services)
])
_FINDERS_INSTALLED = True
def _module_name_to_proto_file(suffix, module_name):
components = module_name.split(".")
proto_name = components[-1][:-1 * len(suffix)]
# NOTE(rbellevi): The Protobuf library expects this path to use
# forward slashes on every platform.
return "/".join(components[:-1] + [proto_name + ".proto"])
def _proto_file_to_module_name(suffix, proto_file):
components = proto_file.split(os.path.sep)
proto_base_name = os.path.splitext(components[-1])[0]
return ".".join(components[:-1] + [proto_base_name + suffix])
def _protos(protobuf_path):
"""Returns a gRPC module generated from the indicated proto file."""
_maybe_install_proto_finders()
module_name = _proto_file_to_module_name(_PROTO_MODULE_SUFFIX,
protobuf_path)
module = importlib.import_module(module_name)
return module
def _services(protobuf_path):
"""Returns a module generated from the indicated proto file."""
_maybe_install_proto_finders()
_protos(protobuf_path)
module_name = _proto_file_to_module_name(_SERVICE_MODULE_SUFFIX,
protobuf_path)
module = importlib.import_module(module_name)
return module
def _protos_and_services(protobuf_path):
"""Returns two modules, corresponding to _pb2.py and _pb2_grpc.py files."""
return (_protos(protobuf_path), _services(protobuf_path))
_proto_code_cache = {}
_proto_code_cache_lock = threading.RLock()
class ProtoLoader(importlib.abc.Loader):
def __init__(self, suffix, codegen_fn, module_name, protobuf_path,
proto_root):
self._suffix = suffix
self._codegen_fn = codegen_fn
self._module_name = module_name
self._protobuf_path = protobuf_path
self._proto_root = proto_root
def create_module(self, spec):
return None
def _generated_file_to_module_name(self, filepath):
components = filepath.split(os.path.sep)
return ".".join(components[:-1] +
[os.path.splitext(components[-1])[0]])
def exec_module(self, module):
assert module.__name__ == self._module_name
code = None
with _proto_code_cache_lock:
if self._module_name in _proto_code_cache:
code = _proto_code_cache[self._module_name]
exec(code, module.__dict__)
else:
files = self._codegen_fn(
self._protobuf_path.encode('ascii'),
[path.encode('ascii') for path in sys.path])
# NOTE: The files are returned in topological order of dependencies. Each
# entry is guaranteed to depend only on the modules preceding it in the
# list and the last entry is guaranteed to be our requested module. We
# cache the code from the first invocation at module-scope so that we
# don't have to regenerate code that has already been generated by protoc.
for f in files[:-1]:
module_name = self._generated_file_to_module_name(
f[0].decode('ascii'))
if module_name not in sys.modules:
if module_name not in _proto_code_cache:
_proto_code_cache[module_name] = f[1]
importlib.import_module(module_name)
exec(files[-1][1], module.__dict__)
class ProtoFinder(importlib.abc.MetaPathFinder):
def __init__(self, suffix, codegen_fn):
self._suffix = suffix
self._codegen_fn = codegen_fn
def find_spec(self, fullname, path, target=None):
if not fullname.endswith(self._suffix):
return None
filepath = _module_name_to_proto_file(self._suffix, fullname)
for search_path in sys.path:
try:
prospective_path = os.path.join(search_path, filepath)
os.stat(prospective_path)
except (FileNotFoundError, NotADirectoryError, OSError):
continue
else:
return importlib.machinery.ModuleSpec(
fullname,
ProtoLoader(self._suffix, self._codegen_fn, fullname,
filepath, search_path))
# NOTE(rbellevi): We provide an environment variable that enables users to completely
# disable this behavior if it is not desired, e.g. for performance reasons.
if not os.getenv(_DISABLE_DYNAMIC_STUBS):
_maybe_install_proto_finders()
if __name__ == '__main__':
proto_include = pkg_resources.resource_filename('grpc_tools', '_proto')
sys.exit(main(sys.argv + ['-I{}'.format(proto_include)]))