diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/grpc_tools/protoc.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/grpc_tools/protoc.py | 210 |
1 files changed, 210 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/grpc_tools/protoc.py b/.venv/lib/python3.12/site-packages/grpc_tools/protoc.py new file mode 100644 index 00000000..2d39d162 --- /dev/null +++ b/.venv/lib/python3.12/site-packages/grpc_tools/protoc.py @@ -0,0 +1,210 @@ +#!/usr/bin/env python3 + +# 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 os +import sys + +from grpc_tools import _protoc_compiler + +if sys.version_info >= (3, 9, 0): + from importlib import resources +else: + import pkg_resources + +_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) + + +def _get_resource_file_name( + package_or_requirement: str, resource_name: str +) -> str: + """Obtain the filename for a resource on the file system.""" + file_name = None + if sys.version_info >= (3, 9, 0): + file_name = ( + resources.files(package_or_requirement) / resource_name + ).resolve() + else: + file_name = pkg_resources.resource_filename( + package_or_requirement, resource_name + ) + return str(file_name) + + +# NOTE(rbellevi): importlib.abc is not supported on 3.4. +if sys.version_info >= (3, 5, 0): + import contextlib + import importlib + import importlib.abc + 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, + ), + ] + ) + + proto_include = _get_resource_file_name("grpc_tools", "_proto") + sys.path.append(proto_include) + + _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 = _get_resource_file_name("grpc_tools", "_proto") + sys.exit(main(sys.argv + ["-I{}".format(proto_include)])) |