Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-21 15:45
📄 __init__.py # Copyright 2015-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. """gRPC's Python API.""" import abc import contextlib import enum import logging import sys from grpc import _compression from grpc._cython import cygrpc as _cygrpc from grpc._runtime_protos import protos from grpc._runtime_protos import protos_and_services from grpc._runtime_protos import services logging.getLogger(__name__).addHandler(logging.NullHandler()) try: # pylint: disable=ungrouped-imports from grpc._grpcio_metadata import __version__ except ImportError: __version__ = "dev0" ############################## Future Interface ############################### class FutureTimeoutError(Exception): """Indicates that a method call on a Future timed out.""" class FutureCancelledError(Exception): """Indicates that the computation underlying a Future was cancelled.""" class Future(abc.ABC): """A representation of a computation in another control flow. Computations represented by a Future may be yet to be begun, may be ongoing, or may have already completed. """ @abc.abstractmethod def cancel(self): """Attempts to cancel the computation. This method does not block. Returns: bool: Returns True if the computation was canceled. Returns False under all other circumstances, for example: 1. computation has begun and could not be canceled. 2. computation has finished 3. computation is scheduled for execution and it is impossible to determine its state without blocking. """ raise NotImplementedError() @abc.abstractmethod def cancelled(self): """Describes whether the computation was cancelled. This method does not block. Returns: bool: Returns True if the computation was cancelled before its result became available. Returns False under all other circumstances, for example: 1. computation was not cancelled. 2. computation's result is available. """ raise NotImplementedError() @abc.abstractmethod def running(self): """Describes whether the computation is taking place. This method does not block. Returns: Returns True if the computation is scheduled for execution or currently executing. Returns False if the computation already executed or was cancelled. """ raise NotImplementedError() @abc.abstractmethod def done(self): """Describes whether the computation has taken place. This method does not block. Returns: bool: Returns True if the computation already executed or was cancelled. Returns False if the computation is scheduled for execution or currently executing. This is exactly opposite of the running() method's result. """ raise NotImplementedError() @abc.abstractmethod def result(self, timeout=None): """Returns the result of the computation or raises its exception. This method may return immediately or may block. Args: timeout: The length of time in seconds to wait for the computation to finish or be cancelled. If None, the call will block until the computations's termination. Returns: The return value of the computation. Raises: FutureTimeoutError: If a timeout value is passed and the computation does not terminate within the allotted time. FutureCancelledError: If the computation was cancelled. Exception: If the computation raised an exception, this call will raise the same exception. """ raise NotImplementedError() @abc.abstractmethod def exception(self, timeout=None): """Return the exception raised by the computation. This method may return immediately or may block. Args: timeout: The length of time in seconds to wait for the computation to terminate or be cancelled. If None, the call will block until the computations's termination. Returns: The exception raised by the computation, or None if the computation did not raise an exception. Raises: FutureTimeoutError: If a timeout value is passed and the computation does not terminate within the allotted time. FutureCancelledError: If the computation was cancelled. """ raise NotImplementedError() @abc.abstractmethod def traceback(self, timeout=None): """Access the traceback of the exception raised by the computation. This method may return immediately or may block. Args: timeout: The length of time in seconds to wait for the computation to terminate or be cancelled. If None, the call will block until the computation's termination. Returns: The traceback of the exception raised by the computation, or None if the computation did not raise an exception. Raises: FutureTimeoutError: If a timeout value is passed and the computation does not terminate within the allotted time. FutureCancelledError: If the computation was cancelled. """ raise NotImplementedError() @abc.abstractmethod def add_done_callback(self, fn): """Adds a function to be called at completion of the computation. The callback will be passed this Future object describing the outcome of the computation. Callbacks will be invoked after the future is terminated, whether successfully or not. If the computation has already completed, the callback will be called immediately. Exceptions raised in the callback will be logged at ERROR level, but will not terminate any threads of execution. Args: fn: A callable taking this Future object as its single parameter. """ raise NotImplementedError() ################################ gRPC Enums ################################## @enum.unique class ChannelConnectivity(enum.Enum): """Mirrors grpc_connectivity_state in the gRPC Core. Attributes: IDLE: The channel is idle. CONNECTING: The channel is connecting. READY: The channel is ready to conduct RPCs. TRANSIENT_FAILURE: The channel has seen a failure from which it expects to recover. SHUTDOWN: The channel has seen a failure from which it cannot recover. """ IDLE = (_cygrpc.ConnectivityState.idle, "idle") CONNECTING = (_cygrpc.ConnectivityState.connecting, "connecting") READY = (_cygrpc.ConnectivityState.ready, "ready") TRANSIENT_FAILURE = ( _cygrpc.ConnectivityState.transient_failure, "transient failure", ) SHUTDOWN = (_cygrpc.ConnectivityState.shutdown, "shutdown") @enum.unique class StatusCode(enum.Enum): """Mirrors grpc_status_code in the gRPC Core. Attributes: OK: Not an error; returned on success CANCELLED: The operation was cancelled (typically by the caller). UNKNOWN: Unknown error. INVALID_ARGUMENT: Client specified an invalid argument. DEADLINE_EXCEEDED: Deadline expired before operation could complete. NOT_FOUND: Some requested entity (e.g., file or directory) was not found. ALREADY_EXISTS: Some entity that we attempted to create (e.g., file or directory) already exists. PERMISSION_DENIED: The caller does not have permission to execute the specified operation. UNAUTHENTICATED: The request does not have valid authentication credentials for the operation. RESOURCE_EXHAUSTED: Some resource has been exhausted, perhaps a per-user quota, or perhaps the entire file system is out of space. FAILED_PRECONDITION: Operation was rejected because the system is not in a state required for the operation's execution. ABORTED: The operation was aborted, typically due to a concurrency issue like sequencer check failures, transaction aborts, etc. UNIMPLEMENTED: Operation is not implemented or not supported/enabled in this service. INTERNAL: Internal errors. Means some invariants expected by underlying system has been broken. UNAVAILABLE: The service is currently unavailable. DATA_LOSS: Unrecoverable data loss or corruption. """ OK = (_cygrpc.StatusCode.ok, "ok") CANCELLED = (_cygrpc.StatusCode.cancelled, "cancelled") UNKNOWN = (_cygrpc.StatusCode.unknown, "unknown") INVALID_ARGUMENT = (_cygrpc.StatusCode.invalid_argument, "invalid argument") DEADLINE_EXCEEDED = ( _cygrpc.StatusCode.deadline_exceeded, "deadline exceeded", ) NOT_FOUND = (_cygrpc.StatusCode.not_found, "not found") ALREADY_EXISTS = (_cygrpc.StatusCode.already_exists, "already exists") PERMISSION_DENIED = ( _cygrpc.StatusCode.permission_denied, "permission denied", ) RESOURCE_EXHAUSTED = ( _cygrpc.StatusCode.resource_exhausted, "resource exhausted", ) FAILED_PRECONDITION = ( _cygrpc.StatusCode.failed_precondition, "failed precondition", ) ABORTED = (_cygrpc.StatusCode.aborted, "aborted") OUT_OF_RANGE = (_cygrpc.StatusCode.out_of_range, "out of range") UNIMPLEMENTED = (_cygrpc.StatusCode.unimplemented, "unimplemented") INTERNAL = (_cygrpc.StatusCode.internal, "internal") UNAVAILABLE = (_cygrpc.StatusCode.unavailable, "unavailable") DATA_LOSS = (_cygrpc.StatusCode.data_loss, "data loss") UNAUTHENTICATED = (_cygrpc.StatusCode.unauthenticated, "unauthenticated") ############################# gRPC Status ################################ class Status(abc.ABC): """Describes the status of an RPC. This is an EXPERIMENTAL API. Attributes: code: A StatusCode object to be sent to the client. details: A UTF-8-encodable string to be sent to the client upon termination of the RPC. trailing_metadata: The trailing :term:`metadata` in the RPC. """ ############################# gRPC Exceptions ################################ class RpcError(Exception): """Raised by the gRPC library to indicate non-OK-status RPC termination.""" ############################## Shared Context ################################ class RpcContext(abc.ABC): """Provides RPC-related information and action.""" @abc.abstractmethod def is_active(self): """Describes whether the RPC is active or has terminated. Returns: bool: True if RPC is active, False otherwise. """ raise NotImplementedError() @abc.abstractmethod def time_remaining(self): """Describes the length of allowed time remaining for the RPC. Returns: A nonnegative float indicating the length of allowed time in seconds remaining for the RPC to complete before it is considered to have timed out, or None if no deadline was specified for the RPC. """ raise NotImplementedError() @abc.abstractmethod def cancel(self): """Cancels the RPC. Idempotent and has no effect if the RPC has already terminated. """ raise NotImplementedError() @abc.abstractmethod def add_callback(self, callback): """Registers a callback to be called on RPC termination. Args: callback: A no-parameter callable to be called on RPC termination. Returns: True if the callback was added and will be called later; False if the callback was not added and will not be called (because the RPC already terminated or some other reason). """ raise NotImplementedError() ######################### Invocation-Side Context ############################ class Call(RpcContext, metaclass=abc.ABCMeta): """Invocation-side utility object for an RPC.""" @abc.abstractmethod def initial_metadata(self): """Accesses the initial metadata sent by the server. This method blocks until the value is available. Returns: The initial :term:`metadata`. """ raise NotImplementedError() @abc.abstractmethod def trailing_metadata(self): """Accesses the trailing metadata sent by the server. This method blocks until the value is available. Returns: The trailing :term:`metadata`. """ raise NotImplementedError() @abc.abstractmethod def code(self): """Accesses the status code sent by the server. This method blocks until the value is available. Returns: The StatusCode value for the RPC. """ raise NotImplementedError() @abc.abstractmethod def details(self): """Accesses the details sent by the server. This method blocks until the value is available. Returns: The details string of the RPC. """ raise NotImplementedError() ############## Invocation-Side Interceptor Interfaces & Classes ############## class ClientCallDetails(abc.ABC): """Describes an RPC to be invoked. Attributes: method: The method name of the RPC. timeout: An optional duration of time in seconds to allow for the RPC. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. """ class UnaryUnaryClientInterceptor(abc.ABC): """Affords intercepting unary-unary invocations.""" @abc.abstractmethod def intercept_unary_unary(self, continuation, client_call_details, request): """Intercepts a unary-unary invocation asynchronously. Args: continuation: A function that proceeds with the invocation by executing the next interceptor in chain or invoking the actual RPC on the underlying Channel. It is the interceptor's responsibility to call it if it decides to move the RPC forward. The interceptor can use `response_future = continuation(client_call_details, request)` to continue with the RPC. `continuation` returns an object that is both a Call for the RPC and a Future. In the event of RPC completion, the return Call-Future's result value will be the response message of the RPC. Should the event terminate with non-OK status, the returned Call-Future's exception value will be an RpcError. client_call_details: A ClientCallDetails object describing the outgoing RPC. request: The request value for the RPC. Returns: An object that is both a Call for the RPC and a Future. In the event of RPC completion, the return Call-Future's result value will be the response message of the RPC. Should the event terminate with non-OK status, the returned Call-Future's exception value will be an RpcError. """ raise NotImplementedError() class UnaryStreamClientInterceptor(abc.ABC): """Affords intercepting unary-stream invocations.""" @abc.abstractmethod def intercept_unary_stream( self, continuation, client_call_details, request ): """Intercepts a unary-stream invocation. Args: continuation: A function that proceeds with the invocation by executing the next interceptor in chain or invoking the actual RPC on the underlying Channel. It is the interceptor's responsibility to call it if it decides to move the RPC forward. The interceptor can use `response_iterator = continuation(client_call_details, request)` to continue with the RPC. `continuation` returns an object that is both a Call for the RPC and an iterator for response values. Drawing response values from the returned Call-iterator may raise RpcError indicating termination of the RPC with non-OK status. client_call_details: A ClientCallDetails object describing the outgoing RPC. request: The request value for the RPC. Returns: An object that is both a Call for the RPC and an iterator of response values. Drawing response values from the returned Call-iterator may raise RpcError indicating termination of the RPC with non-OK status. This object should also fulfill the Future interface, though it may not. """ raise NotImplementedError() class StreamUnaryClientInterceptor(abc.ABC): """Affords intercepting stream-unary invocations.""" @abc.abstractmethod def intercept_stream_unary( self, continuation, client_call_details, request_iterator ): """Intercepts a stream-unary invocation asynchronously. Args: continuation: A function that proceeds with the invocation by executing the next interceptor in chain or invoking the actual RPC on the underlying Channel. It is the interceptor's responsibility to call it if it decides to move the RPC forward. The interceptor can use `response_future = continuation(client_call_details, request_iterator)` to continue with the RPC. `continuation` returns an object that is both a Call for the RPC and a Future. In the event of RPC completion, the return Call-Future's result value will be the response message of the RPC. Should the event terminate with non-OK status, the returned Call-Future's exception value will be an RpcError. client_call_details: A ClientCallDetails object describing the outgoing RPC. request_iterator: An iterator that yields request values for the RPC. Returns: An object that is both a Call for the RPC and a Future. In the event of RPC completion, the return Call-Future's result value will be the response message of the RPC. Should the event terminate with non-OK status, the returned Call-Future's exception value will be an RpcError. """ raise NotImplementedError() class StreamStreamClientInterceptor(abc.ABC): """Affords intercepting stream-stream invocations.""" @abc.abstractmethod def intercept_stream_stream( self, continuation, client_call_details, request_iterator ): """Intercepts a stream-stream invocation. Args: continuation: A function that proceeds with the invocation by executing the next interceptor in chain or invoking the actual RPC on the underlying Channel. It is the interceptor's responsibility to call it if it decides to move the RPC forward. The interceptor can use `response_iterator = continuation(client_call_details, request_iterator)` to continue with the RPC. `continuation` returns an object that is both a Call for the RPC and an iterator for response values. Drawing response values from the returned Call-iterator may raise RpcError indicating termination of the RPC with non-OK status. client_call_details: A ClientCallDetails object describing the outgoing RPC. request_iterator: An iterator that yields request values for the RPC. Returns: An object that is both a Call for the RPC and an iterator of response values. Drawing response values from the returned Call-iterator may raise RpcError indicating termination of the RPC with non-OK status. This object should also fulfill the Future interface, though it may not. """ raise NotImplementedError() ############ Authentication & Authorization Interfaces & Classes ############# class ChannelCredentials(object): """An encapsulation of the data required to create a secure Channel. This class has no supported interface - it exists to define the type of its instances and its instances exist to be passed to other functions. For example, ssl_channel_credentials returns an instance of this class and secure_channel requires an instance of this class. """ def __init__(self, credentials): self._credentials = credentials class CallCredentials(object): """An encapsulation of the data required to assert an identity over a call. A CallCredentials has to be used with secure Channel, otherwise the metadata will not be transmitted to the server. A CallCredentials may be composed with ChannelCredentials to always assert identity for every call over that Channel. This class has no supported interface - it exists to define the type of its instances and its instances exist to be passed to other functions. """ def __init__(self, credentials): self._credentials = credentials class AuthMetadataContext(abc.ABC): """Provides information to call credentials metadata plugins. Attributes: service_url: A string URL of the service being called into. method_name: A string of the fully qualified method name being called. """ class AuthMetadataPluginCallback(abc.ABC): """Callback object received by a metadata plugin.""" def __call__(self, metadata, error): """Passes to the gRPC runtime authentication metadata for an RPC. Args: metadata: The :term:`metadata` used to construct the CallCredentials. error: An Exception to indicate error or None to indicate success. """ raise NotImplementedError() class AuthMetadataPlugin(abc.ABC): """A specification for custom authentication.""" def __call__(self, context, callback): """Implements authentication by passing metadata to a callback. This method will be invoked asynchronously in a separate thread. Args: context: An AuthMetadataContext providing information on the RPC that the plugin is being called to authenticate. callback: An AuthMetadataPluginCallback to be invoked either synchronously or asynchronously. """ raise NotImplementedError() class ServerCredentials(object): """An encapsulation of the data required to open a secure port on a Server. This class has no supported interface - it exists to define the type of its instances and its instances exist to be passed to other functions. """ def __init__(self, credentials): self._credentials = credentials class ServerCertificateConfiguration(object): """A certificate configuration for use with an SSL-enabled Server. Instances of this class can be returned in the certificate configuration fetching callback. This class has no supported interface -- it exists to define the type of its instances and its instances exist to be passed to other functions. """ def __init__(self, certificate_configuration): self._certificate_configuration = certificate_configuration ######################## Multi-Callable Interfaces ########################### class UnaryUnaryMultiCallable(abc.ABC): """Affords invoking a unary-unary RPC from client-side.""" @abc.abstractmethod def __call__( self, request, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None, ): """Synchronously invokes the underlying RPC. Args: request: The request value for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. Returns: The response value for the RPC. Raises: RpcError: Indicating that the RPC terminated with non-OK status. The raised RpcError will also be a Call for the RPC affording the RPC's metadata, status code, and details. """ raise NotImplementedError() @abc.abstractmethod def with_call( self, request, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None, ): """Synchronously invokes the underlying RPC. Args: request: The request value for the RPC. timeout: An optional durating of time in seconds to allow for the RPC. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. Returns: The response value for the RPC and a Call value for the RPC. Raises: RpcError: Indicating that the RPC terminated with non-OK status. The raised RpcError will also be a Call for the RPC affording the RPC's metadata, status code, and details. """ raise NotImplementedError() @abc.abstractmethod def future( self, request, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None, ): """Asynchronously invokes the underlying RPC. Args: request: The request value for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. Returns: An object that is both a Call for the RPC and a Future. In the event of RPC completion, the return Call-Future's result value will be the response message of the RPC. Should the event terminate with non-OK status, the returned Call-Future's exception value will be an RpcError. """ raise NotImplementedError() class UnaryStreamMultiCallable(abc.ABC): """Affords invoking a unary-stream RPC from client-side.""" @abc.abstractmethod def __call__( self, request, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None, ): """Invokes the underlying RPC. Args: request: The request value for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. If None, the timeout is considered infinite. metadata: An optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. Returns: An object that is a Call for the RPC, an iterator of response values, and a Future for the RPC. Drawing response values from the returned Call-iterator may raise RpcError indicating termination of the RPC with non-OK status. """ raise NotImplementedError() class StreamUnaryMultiCallable(abc.ABC): """Affords invoking a stream-unary RPC from client-side.""" @abc.abstractmethod def __call__( self, request_iterator, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None, ): """Synchronously invokes the underlying RPC. Args: request_iterator: An iterator that yields request values for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. If None, the timeout is considered infinite. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. Returns: The response value for the RPC. Raises: RpcError: Indicating that the RPC terminated with non-OK status. The raised RpcError will also implement grpc.Call, affording methods such as metadata, code, and details. """ raise NotImplementedError() @abc.abstractmethod def with_call( self, request_iterator, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None, ): """Synchronously invokes the underlying RPC on the client. Args: request_iterator: An iterator that yields request values for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. If None, the timeout is considered infinite. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. Returns: The response value for the RPC and a Call object for the RPC. Raises: RpcError: Indicating that the RPC terminated with non-OK status. The raised RpcError will also be a Call for the RPC affording the RPC's metadata, status code, and details. """ raise NotImplementedError() @abc.abstractmethod def future( self, request_iterator, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None, ): """Asynchronously invokes the underlying RPC on the client. Args: request_iterator: An iterator that yields request values for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. If None, the timeout is considered infinite. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. Returns: An object that is both a Call for the RPC and a Future. In the event of RPC completion, the return Call-Future's result value will be the response message of the RPC. Should the event terminate with non-OK status, the returned Call-Future's exception value will be an RpcError. """ raise NotImplementedError() class StreamStreamMultiCallable(abc.ABC): """Affords invoking a stream-stream RPC on client-side.""" @abc.abstractmethod def __call__( self, request_iterator, timeout=None, metadata=None, credentials=None, wait_for_ready=None, compression=None, ): """Invokes the underlying RPC on the client. Args: request_iterator: An iterator that yields request values for the RPC. timeout: An optional duration of time in seconds to allow for the RPC. If not specified, the timeout is considered infinite. metadata: Optional :term:`metadata` to be transmitted to the service-side of the RPC. credentials: An optional CallCredentials for the RPC. Only valid for secure Channel. wait_for_ready: An optional flag to enable :term:`wait_for_ready` mechanism. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. Returns: An object that is a Call for the RPC, an iterator of response values, and a Future for the RPC. Drawing response values from the returned Call-iterator may raise RpcError indicating termination of the RPC with non-OK status. """ raise NotImplementedError() ############################# Channel Interface ############################## class Channel(abc.ABC): """Affords RPC invocation via generic methods on client-side. Channel objects implement the Context Manager type, although they need not support being entered and exited multiple times. """ @abc.abstractmethod def subscribe(self, callback, try_to_connect=False): """Subscribe to this Channel's connectivity state machine. A Channel may be in any of the states described by ChannelConnectivity. This method allows application to monitor the state transitions. The typical use case is to debug or gain better visibility into gRPC runtime's state. Args: callback: A callable to be invoked with ChannelConnectivity argument. ChannelConnectivity describes current state of the channel. The callable will be invoked immediately upon subscription and again for every change to ChannelConnectivity until it is unsubscribed or this Channel object goes out of scope. try_to_connect: A boolean indicating whether or not this Channel should attempt to connect immediately. If set to False, gRPC runtime decides when to connect. """ raise NotImplementedError() @abc.abstractmethod def unsubscribe(self, callback): """Unsubscribes a subscribed callback from this Channel's connectivity. Args: callback: A callable previously registered with this Channel from having been passed to its "subscribe" method. """ raise NotImplementedError() @abc.abstractmethod def unary_unary( self, method, request_serializer=None, response_deserializer=None, _registered_method=False, ): """Creates a UnaryUnaryMultiCallable for a unary-unary method. Args: method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. _registered_method: Implementation Private. A bool representing whether the method is registered. Returns: A UnaryUnaryMultiCallable value for the named unary-unary method. """ raise NotImplementedError() @abc.abstractmethod def unary_stream( self, method, request_serializer=None, response_deserializer=None, _registered_method=False, ): """Creates a UnaryStreamMultiCallable for a unary-stream method. Args: method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. _registered_method: Implementation Private. A bool representing whether the method is registered. Returns: A UnaryStreamMultiCallable value for the name unary-stream method. """ raise NotImplementedError() @abc.abstractmethod def stream_unary( self, method, request_serializer=None, response_deserializer=None, _registered_method=False, ): """Creates a StreamUnaryMultiCallable for a stream-unary method. Args: method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. _registered_method: Implementation Private. A bool representing whether the method is registered. Returns: A StreamUnaryMultiCallable value for the named stream-unary method. """ raise NotImplementedError() @abc.abstractmethod def stream_stream( self, method, request_serializer=None, response_deserializer=None, _registered_method=False, ): """Creates a StreamStreamMultiCallable for a stream-stream method. Args: method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. _registered_method: Implementation Private. A bool representing whether the method is registered. Returns: A StreamStreamMultiCallable value for the named stream-stream method. """ raise NotImplementedError() @abc.abstractmethod def close(self): """Closes this Channel and releases all resources held by it. Closing the Channel will immediately terminate all RPCs active with the Channel and it is not valid to invoke new RPCs with the Channel. This method is idempotent. """ raise NotImplementedError() def __enter__(self): """Enters the runtime context related to the channel object.""" raise NotImplementedError() def __exit__(self, exc_type, exc_val, exc_tb): """Exits the runtime context related to the channel object.""" raise NotImplementedError() ########################## Service-Side Context ############################## class ServicerContext(RpcContext, metaclass=abc.ABCMeta): """A context object passed to method implementations.""" @abc.abstractmethod def invocation_metadata(self): """Accesses the metadata sent by the client. Returns: The invocation :term:`metadata`. """ raise NotImplementedError() @abc.abstractmethod def peer(self): """Identifies the peer that invoked the RPC being serviced. Returns: A string identifying the peer that invoked the RPC being serviced. The string format is determined by gRPC runtime. """ raise NotImplementedError() @abc.abstractmethod def peer_identities(self): """Gets one or more peer identity(s). Equivalent to servicer_context.auth_context().get(servicer_context.peer_identity_key()) Returns: An iterable of the identities, or None if the call is not authenticated. Each identity is returned as a raw bytes type. """ raise NotImplementedError() @abc.abstractmethod def peer_identity_key(self): """The auth property used to identify the peer. For example, "x509_common_name" or "x509_subject_alternative_name" are used to identify an SSL peer. Returns: The auth property (string) that indicates the peer identity, or None if the call is not authenticated. """ raise NotImplementedError() @abc.abstractmethod def auth_context(self): """Gets the auth context for the call. Returns: A map of strings to an iterable of bytes for each auth property. """ raise NotImplementedError() def set_compression(self, compression): """Set the compression algorithm to be used for the entire call. Args: compression: An element of grpc.compression, e.g. grpc.compression.Gzip. """ raise NotImplementedError() @abc.abstractmethod def send_initial_metadata(self, initial_metadata): """Sends the initial metadata value to the client. This method need not be called by implementations if they have no metadata to add to what the gRPC runtime will transmit. Args: initial_metadata: The initial :term:`metadata`. """ raise NotImplementedError() @abc.abstractmethod def set_trailing_metadata(self, trailing_metadata): """Sets the trailing metadata for the RPC. Sets the trailing metadata to be sent upon completion of the RPC. If this method is invoked multiple times throughout the lifetime of an RPC, the value supplied in the final invocation will be the value sent over the wire. This method need not be called by implementations if they have no metadata to add to what the gRPC runtime will transmit. Args: trailing_metadata: The trailing :term:`metadata`. """ raise NotImplementedError() def trailing_metadata(self): """Access value to be used as trailing metadata upon RPC completion. This is an EXPERIMENTAL API. Returns: The trailing :term:`metadata` for the RPC. """ raise NotImplementedError() @abc.abstractmethod def abort(self, code, details): """Raises an exception to terminate the RPC with a non-OK status. The code and details passed as arguments will supersede any existing ones. Args: code: A StatusCode object to be sent to the client. It must not be StatusCode.OK. details: A UTF-8-encodable string to be sent to the client upon termination of the RPC. Raises: Exception: An exception is always raised to signal the abortion the RPC to the gRPC runtime. """ raise NotImplementedError() @abc.abstractmethod def abort_with_status(self, status): """Raises an exception to terminate the RPC with a non-OK status. The status passed as argument will supersede any existing status code, status message and trailing metadata. This is an EXPERIMENTAL API. Args: status: A grpc.Status object. The status code in it must not be StatusCode.OK. Raises: Exception: An exception is always raised to signal the abortion the RPC to the gRPC runtime. """ raise NotImplementedError() @abc.abstractmethod def set_code(self, code): """Sets the value to be used as status code upon RPC completion. This method need not be called by method implementations if they wish the gRPC runtime to determine the status code of the RPC. Args: code: A StatusCode object to be sent to the client. """ raise NotImplementedError() @abc.abstractmethod def set_details(self, details): """Sets the value to be used as detail string upon RPC completion. This method need not be called by method implementations if they have no details to transmit. Args: details: A UTF-8-encodable string to be sent to the client upon termination of the RPC. """ raise NotImplementedError() def code(self): """Accesses the value to be used as status code upon RPC completion. This is an EXPERIMENTAL API. Returns: The StatusCode value for the RPC. """ raise NotImplementedError() def details(self): """Accesses the value to be used as detail string upon RPC completion. This is an EXPERIMENTAL API. Returns: The details string of the RPC. """ raise NotImplementedError() def disable_next_message_compression(self): """Disables compression for the next response message. This method will override any compression configuration set during server creation or set on the call. """ raise NotImplementedError() ##################### Service-Side Handler Interfaces ######################## class RpcMethodHandler(abc.ABC): """An implementation of a single RPC method. Attributes: request_streaming: Whether the RPC supports exactly one request message or any arbitrary number of request messages. response_streaming: Whether the RPC supports exactly one response message or any arbitrary number of response messages. request_deserializer: A callable :term:`deserializer` that accepts a byte string and returns an object suitable to be passed to this object's business logic, or None to indicate that this object's business logic should be passed the raw request bytes. response_serializer: A callable :term:`serializer` that accepts an object produced by this object's business logic and returns a byte string, or None to indicate that the byte strings produced by this object's business logic should be transmitted on the wire as they are. unary_unary: This object's application-specific business logic as a callable value that takes a request value and a ServicerContext object and returns a response value. Only non-None if both request_streaming and response_streaming are False. unary_stream: This object's application-specific business logic as a callable value that takes a request value and a ServicerContext object and returns an iterator of response values. Only non-None if request_streaming is False and response_streaming is True. stream_unary: This object's application-specific business logic as a callable value that takes an iterator of request values and a ServicerContext object and returns a response value. Only non-None if request_streaming is True and response_streaming is False. stream_stream: This object's application-specific business logic as a callable value that takes an iterator of request values and a ServicerContext object and returns an iterator of response values. Only non-None if request_streaming and response_streaming are both True. """ class HandlerCallDetails(abc.ABC): """Describes an RPC that has just arrived for service. Attributes: method: The method name of the RPC. invocation_metadata: The :term:`metadata` sent by the client. """ class GenericRpcHandler(abc.ABC): """An implementation of arbitrarily many RPC methods.""" @abc.abstractmethod def service(self, handler_call_details): """Returns the handler for servicing the RPC. Args: handler_call_details: A HandlerCallDetails describing the RPC. Returns: An RpcMethodHandler with which the RPC may be serviced if the implementation chooses to service this RPC, or None otherwise. """ raise NotImplementedError() class ServiceRpcHandler(GenericRpcHandler, metaclass=abc.ABCMeta): """An implementation of RPC methods belonging to a service. A service handles RPC methods with structured names of the form '/Service.Name/Service.Method', where 'Service.Name' is the value returned by service_name(), and 'Service.Method' is the method name. A service can have multiple method names, but only a single service name. """ @abc.abstractmethod def service_name(self): """Returns this service's name. Returns: The service name. """ raise NotImplementedError() #################### Service-Side Interceptor Interfaces ##################### class ServerInterceptor(abc.ABC): """Affords intercepting incoming RPCs on the service-side.""" @abc.abstractmethod def intercept_service(self, continuation, handler_call_details): """Intercepts incoming RPCs before handing them over to a handler. State can be passed from an interceptor to downstream interceptors via contextvars. The first interceptor is called from an empty contextvars.Context, and the same Context is used for downstream interceptors and for the final handler call. Note that there are no guarantees that interceptors and handlers will be called from the same thread. Args: continuation: A function that takes a HandlerCallDetails and proceeds to invoke the next interceptor in the chain, if any, or the RPC handler lookup logic, with the call details passed as an argument, and returns an RpcMethodHandler instance if the RPC is considered serviced, or None otherwise. handler_call_details: A HandlerCallDetails describing the RPC. Returns: An RpcMethodHandler with which the RPC may be serviced if the interceptor chooses to service this RPC, or None otherwise. """ raise NotImplementedError() ############################# Server Interface ############################### class Server(abc.ABC): """Services RPCs.""" @abc.abstractmethod def add_generic_rpc_handlers(self, generic_rpc_handlers): """Registers GenericRpcHandlers with this Server. This method is only safe to call before the server is started. Args: generic_rpc_handlers: An iterable of GenericRpcHandlers that will be used to service RPCs. """ raise NotImplementedError() def add_registered_method_handlers(self, service_name, method_handlers): """Registers GenericRpcHandlers with this Server. This method is only safe to call before the server is started. If the same method have both generic and registered handler, registered handler will take precedence. Args: service_name: The service name. method_handlers: A dictionary that maps method names to corresponding RpcMethodHandler. """ @abc.abstractmethod def add_insecure_port(self, address): """Opens an insecure port for accepting RPCs. This method may only be called before starting the server. Args: address: The address for which to open a port. If the port is 0, or not specified in the address, then gRPC runtime will choose a port. Returns: An integer port on which server will accept RPC requests. """ raise NotImplementedError() @abc.abstractmethod def add_secure_port(self, address, server_credentials): """Opens a secure port for accepting RPCs. This method may only be called before starting the server. Args: address: The address for which to open a port. if the port is 0, or not specified in the address, then gRPC runtime will choose a port. server_credentials: A ServerCredentials object. Returns: An integer port on which server will accept RPC requests. """ raise NotImplementedError() @abc.abstractmethod def start(self): """Starts this Server. This method may only be called once. (i.e. it is not idempotent). """ raise NotImplementedError() @abc.abstractmethod def stop(self, grace): """Stops this Server. This method immediately stop service of new RPCs in all cases. If a grace period is specified, this method waits until all active RPCs are finished or until the grace period is reached. RPCs that haven't been terminated within the grace period are aborted. If a grace period is not specified (by passing None for `grace`), all existing RPCs are aborted immediately and this method blocks until the last RPC handler terminates. This method is idempotent and may be called at any time. Passing a smaller grace value in a subsequent call will have the effect of stopping the Server sooner (passing None will have the effect of stopping the server immediately). Passing a larger grace value in a subsequent call will not have the effect of stopping the server later (i.e. the most restrictive grace value is used). Args: grace: A duration of time in seconds or None. Returns: A threading.Event that will be set when this Server has completely stopped, i.e. when running RPCs either complete or are aborted and all handlers have terminated. """ raise NotImplementedError() def wait_for_termination(self, timeout=None): """Block current thread until the server stops. This is an EXPERIMENTAL API. The wait will not consume computational resources during blocking, and it will block until one of the two following conditions are met: 1) The server is stopped or terminated; 2) A timeout occurs if timeout is not `None`. The timeout argument works in the same way as `threading.Event.wait()`. https://docs.python.org/3/library/threading.html#threading.Event.wait Args: timeout: A floating point number specifying a timeout for the operation in seconds. Returns: A bool indicates if the operation times out. """ raise NotImplementedError() ################################# Functions ################################ def unary_unary_rpc_method_handler( behavior, request_deserializer=None, response_serializer=None ): """Creates an RpcMethodHandler for a unary-unary RPC method. Args: behavior: The implementation of an RPC that accepts one request and returns one response. request_deserializer: An optional :term:`deserializer` for request deserialization. response_serializer: An optional :term:`serializer` for response serialization. Returns: An RpcMethodHandler object that is typically used by grpc.Server. """ from grpc import _utilities # pylint: disable=cyclic-import return _utilities.RpcMethodHandler( False, False, request_deserializer, response_serializer, behavior, None, None, None, ) def unary_stream_rpc_method_handler( behavior, request_deserializer=None, response_serializer=None ): """Creates an RpcMethodHandler for a unary-stream RPC method. Args: behavior: The implementation of an RPC that accepts one request and returns an iterator of response values. request_deserializer: An optional :term:`deserializer` for request deserialization. response_serializer: An optional :term:`serializer` for response serialization. Returns: An RpcMethodHandler object that is typically used by grpc.Server. """ from grpc import _utilities # pylint: disable=cyclic-import return _utilities.RpcMethodHandler( False, True, request_deserializer, response_serializer, None, behavior, None, None, ) def stream_unary_rpc_method_handler( behavior, request_deserializer=None, response_serializer=None ): """Creates an RpcMethodHandler for a stream-unary RPC method. Args: behavior: The implementation of an RPC that accepts an iterator of request values and returns a single response value. request_deserializer: An optional :term:`deserializer` for request deserialization. response_serializer: An optional :term:`serializer` for response serialization. Returns: An RpcMethodHandler object that is typically used by grpc.Server. """ from grpc import _utilities # pylint: disable=cyclic-import return _utilities.RpcMethodHandler( True, False, request_deserializer, response_serializer, None, None, behavior, None, ) def stream_stream_rpc_method_handler( behavior, request_deserializer=None, response_serializer=None ): """Creates an RpcMethodHandler for a stream-stream RPC method. Args: behavior: The implementation of an RPC that accepts an iterator of request values and returns an iterator of response values. request_deserializer: An optional :term:`deserializer` for request deserialization. response_serializer: An optional :term:`serializer` for response serialization. Returns: An RpcMethodHandler object that is typically used by grpc.Server. """ from grpc import _utilities # pylint: disable=cyclic-import return _utilities.RpcMethodHandler( True, True, request_deserializer, response_serializer, None, None, None, behavior, ) def method_handlers_generic_handler(service, method_handlers): """Creates a GenericRpcHandler from RpcMethodHandlers. Args: service: The name of the service that is implemented by the method_handlers. method_handlers: A dictionary that maps method names to corresponding RpcMethodHandler. Returns: A GenericRpcHandler. This is typically added to the grpc.Server object with add_generic_rpc_handlers() before starting the server. """ from grpc import _utilities # pylint: disable=cyclic-import return _utilities.DictionaryGenericHandler(service, method_handlers) def ssl_channel_credentials( root_certificates=None, private_key=None, certificate_chain=None ): """Creates a ChannelCredentials for use with an SSL-enabled Channel. Args: root_certificates: The PEM-encoded root certificates as a byte string, or None to retrieve them from a default location chosen by gRPC runtime. private_key: The PEM-encoded private key as a byte string, or None if no private key should be used. certificate_chain: The PEM-encoded certificate chain as a byte string to use or None if no certificate chain should be used. Returns: A ChannelCredentials for use with an SSL-enabled Channel. """ return ChannelCredentials( _cygrpc.SSLChannelCredentials( root_certificates, private_key, certificate_chain ) ) def xds_channel_credentials(fallback_credentials=None): """Creates a ChannelCredentials for use with xDS. This is an EXPERIMENTAL API. Args: fallback_credentials: Credentials to use in case it is not possible to establish a secure connection via xDS. If no fallback_credentials argument is supplied, a default SSLChannelCredentials is used. """ fallback_credentials = ( ssl_channel_credentials() if fallback_credentials is None else fallback_credentials ) return ChannelCredentials( _cygrpc.XDSChannelCredentials(fallback_credentials._credentials) ) def metadata_call_credentials(metadata_plugin, name=None): """Construct CallCredentials from an AuthMetadataPlugin. Args: metadata_plugin: An AuthMetadataPlugin to use for authentication. name: An optional name for the plugin. Returns: A CallCredentials. """ from grpc import _plugin_wrapping # pylint: disable=cyclic-import return _plugin_wrapping.metadata_plugin_call_credentials( metadata_plugin, name ) def access_token_call_credentials(access_token): """Construct CallCredentials from an access token. Args: access_token: A string to place directly in the http request authorization header, for example "authorization: Bearer <access_token>". Returns: A CallCredentials. """ from grpc import _auth # pylint: disable=cyclic-import from grpc import _plugin_wrapping # pylint: disable=cyclic-import return _plugin_wrapping.metadata_plugin_call_credentials( _auth.AccessTokenAuthMetadataPlugin(access_token), None ) def composite_call_credentials(call_credentials): """Compose multiple CallCredentials to make a new CallCredentials. Args: call_credentials: At least two CallCredentials objects. Returns: A CallCredentials object composed of the given CallCredentials objects. """ return CallCredentials( _cygrpc.CompositeCallCredentials( tuple( single_call_credentials._credentials for single_call_credentials in call_credentials ) ) ) def composite_channel_credentials(channel_credentials, call_credentials): """Compose a ChannelCredentials and one or more CallCredentials objects. Args: channel_credentials: A ChannelCredentials object. call_credentials: One or more CallCredentials objects. Returns: A ChannelCredentials composed of the given ChannelCredentials and CallCredentials objects. """ return ChannelCredentials( _cygrpc.CompositeChannelCredentials( tuple( single_call_credentials._credentials for single_call_credentials in call_credentials ), channel_credentials._credentials, ) ) def ssl_server_credentials( private_key_certificate_chain_pairs, root_certificates=None, require_client_auth=False, ): """Creates a ServerCredentials for use with an SSL-enabled Server. Args: private_key_certificate_chain_pairs: A list of pairs of the form [PEM-encoded private key, PEM-encoded certificate chain]. root_certificates: An optional byte string of PEM-encoded client root certificates that the server will use to verify client authentication. If omitted, require_client_auth must also be False. require_client_auth: A boolean indicating whether or not to require clients to be authenticated. May only be True if root_certificates is not None. Returns: A ServerCredentials for use with an SSL-enabled Server. Typically, this object is an argument to add_secure_port() method during server setup. """ if not private_key_certificate_chain_pairs: raise ValueError( "At least one private key-certificate chain pair is required!" ) elif require_client_auth and root_certificates is None: raise ValueError( "Illegal to require client auth without providing root" " certificates!" ) else: return ServerCredentials( _cygrpc.server_credentials_ssl( root_certificates, [ _cygrpc.SslPemKeyCertPair(key, pem) for key, pem in private_key_certificate_chain_pairs ], require_client_auth, ) ) def xds_server_credentials(fallback_credentials): """Creates a ServerCredentials for use with xDS. This is an EXPERIMENTAL API. Args: fallback_credentials: Credentials to use in case it is not possible to establish a secure connection via xDS. No default value is provided. """ return ServerCredentials( _cygrpc.xds_server_credentials(fallback_credentials._credentials) ) def insecure_server_credentials(): """Creates a credentials object directing the server to use no credentials. This is an EXPERIMENTAL API. This object cannot be used directly in a call to `add_secure_port`. Instead, it should be used to construct other credentials objects, e.g. with xds_server_credentials. """ return ServerCredentials(_cygrpc.insecure_server_credentials()) def ssl_server_certificate_configuration( private_key_certificate_chain_pairs, root_certificates=None ): """Creates a ServerCertificateConfiguration for use with a Server. Args: private_key_certificate_chain_pairs: A collection of pairs of the form [PEM-encoded private key, PEM-encoded certificate chain]. root_certificates: An optional byte string of PEM-encoded client root certificates that the server will use to verify client authentication. Returns: A ServerCertificateConfiguration that can be returned in the certificate configuration fetching callback. """ if private_key_certificate_chain_pairs: return ServerCertificateConfiguration( _cygrpc.server_certificate_config_ssl( root_certificates, [ _cygrpc.SslPemKeyCertPair(key, pem) for key, pem in private_key_certificate_chain_pairs ], ) ) else: raise ValueError( "At least one private key-certificate chain pair is required!" ) def dynamic_ssl_server_credentials( initial_certificate_configuration, certificate_configuration_fetcher, require_client_authentication=False, ): """Creates a ServerCredentials for use with an SSL-enabled Server. Args: initial_certificate_configuration (ServerCertificateConfiguration): The certificate configuration with which the server will be initialized. certificate_configuration_fetcher (callable): A callable that takes no arguments and should return a ServerCertificateConfiguration to replace the server's current certificate, or None for no change (i.e., the server will continue its current certificate config). The library will call this callback on every new client connection before starting the TLS handshake with the client, thus allowing the user application to optionally return a new ServerCertificateConfiguration that the server will then use for the handshake. require_client_authentication: A boolean indicating whether or not to require clients to be authenticated. Returns: A ServerCredentials. """ return ServerCredentials( _cygrpc.server_credentials_ssl_dynamic_cert_config( initial_certificate_configuration, certificate_configuration_fetcher, require_client_authentication, ) ) @enum.unique class LocalConnectionType(enum.Enum): """Types of local connection for local credential creation. Attributes: UDS: Unix domain socket connections LOCAL_TCP: Local TCP connections. """ UDS = _cygrpc.LocalConnectionType.uds LOCAL_TCP = _cygrpc.LocalConnectionType.local_tcp def local_channel_credentials(local_connect_type=LocalConnectionType.LOCAL_TCP): """Creates a local ChannelCredentials used for local connections. This is an EXPERIMENTAL API. Local credentials are used by local TCP endpoints (e.g. localhost:10000) also UDS connections. The connections created by local channel credentials are not encrypted, but will be checked if they are local or not. The UDS connections are considered secure by providing peer authentication and data confidentiality while TCP connections are considered insecure. It is allowed to transmit call credentials over connections created by local channel credentials. Local channel credentials are useful for 1) eliminating insecure_channel usage; 2) enable unit testing for call credentials without setting up secrets. Args: local_connect_type: Local connection type (either grpc.LocalConnectionType.UDS or grpc.LocalConnectionType.LOCAL_TCP) Returns: A ChannelCredentials for use with a local Channel """ return ChannelCredentials( _cygrpc.channel_credentials_local(local_connect_type.value) ) def local_server_credentials(local_connect_type=LocalConnectionType.LOCAL_TCP): """Creates a local ServerCredentials used for local connections. This is an EXPERIMENTAL API. Local credentials are used by local TCP endpoints (e.g. localhost:10000) also UDS connections. The connections created by local server credentials are not encrypted, but will be checked if they are local or not. The UDS connections are considered secure by providing peer authentication and data confidentiality while TCP connections are considered insecure. It is allowed to transmit call credentials over connections created by local server credentials. Local server credentials are useful for 1) eliminating insecure_channel usage; 2) enable unit testing for call credentials without setting up secrets. Args: local_connect_type: Local connection type (either grpc.LocalConnectionType.UDS or grpc.LocalConnectionType.LOCAL_TCP) Returns: A ServerCredentials for use with a local Server """ return ServerCredentials( _cygrpc.server_credentials_local(local_connect_type.value) ) def alts_channel_credentials(service_accounts=None): """Creates a ChannelCredentials for use with an ALTS-enabled Channel. This is an EXPERIMENTAL API. ALTS credentials API can only be used in GCP environment as it relies on handshaker service being available. For more info about ALTS see https://cloud.google.com/security/encryption-in-transit/application-layer-transport-security Args: service_accounts: A list of server identities accepted by the client. If target service accounts are provided and none of them matches the peer identity of the server, handshake will fail. The arg can be empty if the client does not have any information about trusted server identity. Returns: A ChannelCredentials for use with an ALTS-enabled Channel """ return ChannelCredentials( _cygrpc.channel_credentials_alts(service_accounts or []) ) def alts_server_credentials(): """Creates a ServerCredentials for use with an ALTS-enabled connection. This is an EXPERIMENTAL API. ALTS credentials API can only be used in GCP environment as it relies on handshaker service being available. For more info about ALTS see https://cloud.google.com/security/encryption-in-transit/application-layer-transport-security Returns: A ServerCredentials for use with an ALTS-enabled Server """ return ServerCredentials(_cygrpc.server_credentials_alts()) def compute_engine_channel_credentials(call_credentials): """Creates a compute engine channel credential. This credential can only be used in a GCP environment as it relies on a handshaker service. For more info about ALTS, see https://cloud.google.com/security/encryption-in-transit/application-layer-transport-security This channel credential is expected to be used as part of a composite credential in conjunction with a call credentials that authenticates the VM's default service account. If used with any other sort of call credential, the connection may suddenly and unexpectedly begin failing RPCs. """ return ChannelCredentials( _cygrpc.channel_credentials_compute_engine( call_credentials._credentials ) ) def channel_ready_future(channel): """Creates a Future that tracks when a Channel is ready. Cancelling the Future does not affect the channel's state machine. It merely decouples the Future from channel state machine. Args: channel: A Channel object. Returns: A Future object that matures when the channel connectivity is ChannelConnectivity.READY. """ from grpc import _utilities # pylint: disable=cyclic-import return _utilities.channel_ready_future(channel) def insecure_channel(target, options=None, compression=None): """Creates an insecure Channel to a server. The returned Channel is thread-safe. Args: target: The server address options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC Core runtime) to configure the channel. compression: An optional value indicating the compression method to be used over the lifetime of the channel. Returns: A Channel. """ from grpc import _channel # pylint: disable=cyclic-import return _channel.Channel( target, () if options is None else options, None, compression ) def secure_channel(target, credentials, options=None, compression=None): """Creates a secure Channel to a server. The returned Channel is thread-safe. Args: target: The server address. credentials: A ChannelCredentials instance. options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC Core runtime) to configure the channel. compression: An optional value indicating the compression method to be used over the lifetime of the channel. Returns: A Channel. """ from grpc import _channel # pylint: disable=cyclic-import from grpc.experimental import _insecure_channel_credentials if credentials._credentials is _insecure_channel_credentials: raise ValueError( "secure_channel cannot be called with insecure credentials." + " Call insecure_channel instead." ) return _channel.Channel( target, () if options is None else options, credentials._credentials, compression, ) def intercept_channel(channel, interceptors): """Intercepts a channel through a set of interceptors. Args: channel: A Channel. interceptors: Zero or more objects of type UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor, StreamUnaryClientInterceptor, or StreamStreamClientInterceptor. Interceptors are given control in the order they are listed. Returns: A Channel that intercepts each invocation via the provided interceptors. Raises: TypeError: If interceptor does not derive from any of UnaryUnaryClientInterceptor, UnaryStreamClientInterceptor, StreamUnaryClientInterceptor, or StreamStreamClientInterceptor. """ from grpc import _interceptor # pylint: disable=cyclic-import return _interceptor.intercept_channel(channel, interceptors) def server( thread_pool, handlers=None, interceptors=None, options=None, maximum_concurrent_rpcs=None, compression=None, xds=False, ): """Creates a Server with which RPCs can be serviced. Args: thread_pool: A futures.ThreadPoolExecutor to be used by the Server to execute RPC handlers. handlers: An optional list of GenericRpcHandlers used for executing RPCs. More handlers may be added by calling add_generic_rpc_handlers any time before the server is started. interceptors: An optional list of ServerInterceptor objects that observe and optionally manipulate the incoming RPCs before handing them over to handlers. The interceptors are given control in the order they are specified. This is an EXPERIMENTAL API. options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC runtime) to configure the channel. maximum_concurrent_rpcs: The maximum number of concurrent RPCs this server will service before returning RESOURCE_EXHAUSTED status, or None to indicate no limit. compression: An element of grpc.compression, e.g. grpc.compression.Gzip. This compression algorithm will be used for the lifetime of the server unless overridden. xds: If set to true, retrieves server configuration via xDS. This is an EXPERIMENTAL option. Returns: A Server object. """ from grpc import _server # pylint: disable=cyclic-import return _server.create_server( thread_pool, () if handlers is None else handlers, () if interceptors is None else interceptors, () if options is None else options, maximum_concurrent_rpcs, compression, xds, ) @contextlib.contextmanager def _create_servicer_context(rpc_event, state, request_deserializer): from grpc import _server # pylint: disable=cyclic-import context = _server._Context(rpc_event, state, request_deserializer) yield context context._finalize_state() # pylint: disable=protected-access @enum.unique class Compression(enum.IntEnum): """Indicates the compression method to be used for an RPC. Attributes: NoCompression: Do not use compression algorithm. Deflate: Use "Deflate" compression algorithm. Gzip: Use "Gzip" compression algorithm. """ NoCompression = _compression.NoCompression Deflate = _compression.Deflate Gzip = _compression.Gzip ################################### __all__ ################################# __all__ = ( "FutureTimeoutError", "FutureCancelledError", "Future", "ChannelConnectivity", "StatusCode", "Status", "RpcError", "RpcContext", "Call", "ChannelCredentials", "CallCredentials", "AuthMetadataContext", "AuthMetadataPluginCallback", "AuthMetadataPlugin", "Compression", "ClientCallDetails", "ServerCertificateConfiguration", "ServerCredentials", "LocalConnectionType", "UnaryUnaryMultiCallable", "UnaryStreamMultiCallable", "StreamUnaryMultiCallable", "StreamStreamMultiCallable", "UnaryUnaryClientInterceptor", "UnaryStreamClientInterceptor", "StreamUnaryClientInterceptor", "StreamStreamClientInterceptor", "Channel", "ServicerContext", "RpcMethodHandler", "HandlerCallDetails", "GenericRpcHandler", "ServiceRpcHandler", "Server", "ServerInterceptor", "unary_unary_rpc_method_handler", "unary_stream_rpc_method_handler", "stream_unary_rpc_method_handler", "stream_stream_rpc_method_handler", "method_handlers_generic_handler", "ssl_channel_credentials", "metadata_call_credentials", "access_token_call_credentials", "composite_call_credentials", "composite_channel_credentials", "compute_engine_channel_credentials", "local_channel_credentials", "local_server_credentials", "alts_channel_credentials", "alts_server_credentials", "ssl_server_credentials", "ssl_server_certificate_configuration", "dynamic_ssl_server_credentials", "channel_ready_future", "insecure_channel", "secure_channel", "intercept_channel", "server", "protos", "services", "protos_and_services", "xds_channel_credentials", "xds_server_credentials", "insecure_server_credentials", ) ############################### Extension Shims ################################ # Here to maintain backwards compatibility; avoid using these in new code! try: import grpc_tools sys.modules.update({"grpc.tools": grpc_tools}) except ImportError: pass try: import grpc_health sys.modules.update({"grpc.health": grpc_health}) except ImportError: pass try: import grpc_reflection sys.modules.update({"grpc.reflection": grpc_reflection}) except ImportError: pass # Prevents import order issue in the case of renamed path. if sys.version_info >= (3, 6) and __name__ == "grpc": from grpc import aio # pylint: disable=ungrouped-imports sys.modules.update({"grpc.aio": aio})	80.4 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 _auth.py # 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. """GRPCAuthMetadataPlugins for standard authentication.""" import inspect from typing import Any, Optional import grpc def _sign_request( callback: grpc.AuthMetadataPluginCallback, token: Optional[str], error: Optional[Exception], ): metadata = (("authorization", "Bearer {}".format(token)),) callback(metadata, error) class GoogleCallCredentials(grpc.AuthMetadataPlugin): """Metadata wrapper for GoogleCredentials from the oauth2client library.""" _is_jwt: bool _credentials: Any # TODO(xuanwn): Give credentials an actual type. def __init__(self, credentials: Any): self._credentials = credentials # Hack to determine if these are JWT creds and we need to pass # additional_claims when getting a token self._is_jwt = ( "additional_claims" in inspect.getfullargspec(credentials.get_access_token).args ) def __call__( self, context: grpc.AuthMetadataContext, callback: grpc.AuthMetadataPluginCallback, ): try: if self._is_jwt: access_token = self._credentials.get_access_token( additional_claims={ "aud": context.service_url # pytype: disable=attribute-error } ).access_token else: access_token = self._credentials.get_access_token().access_token except Exception as exception: # pylint: disable=broad-except _sign_request(callback, None, exception) else: _sign_request(callback, access_token, None) class AccessTokenAuthMetadataPlugin(grpc.AuthMetadataPlugin): """Metadata wrapper for raw access token credentials.""" _access_token: str def __init__(self, access_token: str): self._access_token = access_token def __call__( self, context: grpc.AuthMetadataContext, callback: grpc.AuthMetadataPluginCallback, ): _sign_request(callback, self._access_token, None)	2.57 KB	2025-05-25 12:26
📄 _channel.py # 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. """Invocation-side implementation of gRPC Python.""" import copy import functools import logging import os import sys import threading import time import types from typing import ( Any, Callable, Dict, Iterator, List, Optional, Sequence, Set, Tuple, Union, ) import grpc # pytype: disable=pyi-error from grpc import _common # pytype: disable=pyi-error from grpc import _compression # pytype: disable=pyi-error from grpc import _grpcio_metadata # pytype: disable=pyi-error from grpc import _observability # pytype: disable=pyi-error from grpc._cython import cygrpc from grpc._typing import ChannelArgumentType from grpc._typing import DeserializingFunction from grpc._typing import IntegratedCallFactory from grpc._typing import MetadataType from grpc._typing import NullaryCallbackType from grpc._typing import ResponseType from grpc._typing import SerializingFunction from grpc._typing import UserTag import grpc.experimental # pytype: disable=pyi-error _LOGGER = logging.getLogger(__name__) _USER_AGENT = "grpc-python/{}".format(_grpcio_metadata.__version__) _EMPTY_FLAGS = 0 # NOTE(rbellevi): No guarantees are given about the maintenance of this # environment variable. _DEFAULT_SINGLE_THREADED_UNARY_STREAM = ( os.getenv("GRPC_SINGLE_THREADED_UNARY_STREAM") is not None ) _UNARY_UNARY_INITIAL_DUE = ( cygrpc.OperationType.send_initial_metadata, cygrpc.OperationType.send_message, cygrpc.OperationType.send_close_from_client, cygrpc.OperationType.receive_initial_metadata, cygrpc.OperationType.receive_message, cygrpc.OperationType.receive_status_on_client, ) _UNARY_STREAM_INITIAL_DUE = ( cygrpc.OperationType.send_initial_metadata, cygrpc.OperationType.send_message, cygrpc.OperationType.send_close_from_client, cygrpc.OperationType.receive_initial_metadata, cygrpc.OperationType.receive_status_on_client, ) _STREAM_UNARY_INITIAL_DUE = ( cygrpc.OperationType.send_initial_metadata, cygrpc.OperationType.receive_initial_metadata, cygrpc.OperationType.receive_message, cygrpc.OperationType.receive_status_on_client, ) _STREAM_STREAM_INITIAL_DUE = ( cygrpc.OperationType.send_initial_metadata, cygrpc.OperationType.receive_initial_metadata, cygrpc.OperationType.receive_status_on_client, ) _CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE = ( "Exception calling channel subscription callback!" ) _OK_RENDEZVOUS_REPR_FORMAT = ( '<{} of RPC that terminated with:\n\tstatus = {}\n\tdetails = "{}"\n>' ) _NON_OK_RENDEZVOUS_REPR_FORMAT = ( "<{} of RPC that terminated with:\n" "\tstatus = {}\n" '\tdetails = "{}"\n' '\tdebug_error_string = "{}"\n' ">" ) def _deadline(timeout: Optional[float]) -> Optional[float]: return None if timeout is None else time.time() + timeout def _unknown_code_details( unknown_cygrpc_code: Optional[grpc.StatusCode], details: Optional[str] ) -> str: return 'Server sent unknown code {} and details "{}"'.format( unknown_cygrpc_code, details ) class _RPCState(object): condition: threading.Condition due: Set[cygrpc.OperationType] initial_metadata: Optional[MetadataType] response: Any trailing_metadata: Optional[MetadataType] code: Optional[grpc.StatusCode] details: Optional[str] debug_error_string: Optional[str] cancelled: bool callbacks: List[NullaryCallbackType] fork_epoch: Optional[int] rpc_start_time: Optional[float] # In relative seconds rpc_end_time: Optional[float] # In relative seconds method: Optional[str] target: Optional[str] def __init__( self, due: Sequence[cygrpc.OperationType], initial_metadata: Optional[MetadataType], trailing_metadata: Optional[MetadataType], code: Optional[grpc.StatusCode], details: Optional[str], ): # `condition` guards all members of _RPCState. `notify_all` is called on # `condition` when the state of the RPC has changed. self.condition = threading.Condition() # The cygrpc.OperationType objects representing events due from the RPC's # completion queue. If an operation is in `due`, it is guaranteed that # `operate()` has been called on a corresponding operation. But the # converse is not true. That is, in the case of failed `operate()` # calls, there may briefly be events in `due` that do not correspond to # operations submitted to Core. self.due = set(due) self.initial_metadata = initial_metadata self.response = None self.trailing_metadata = trailing_metadata self.code = code self.details = details self.debug_error_string = None # The following three fields are used for observability. # Updates to those fields do not trigger self.condition. self.rpc_start_time = None self.rpc_end_time = None self.method = None self.target = None # The semantics of grpc.Future.cancel and grpc.Future.cancelled are # slightly wonky, so they have to be tracked separately from the rest of the # result of the RPC. This field tracks whether cancellation was requested # prior to termination of the RPC. self.cancelled = False self.callbacks = [] self.fork_epoch = cygrpc.get_fork_epoch() def reset_postfork_child(self): self.condition = threading.Condition() def _abort(state: _RPCState, code: grpc.StatusCode, details: str) -> None: if state.code is None: state.code = code state.details = details if state.initial_metadata is None: state.initial_metadata = () state.trailing_metadata = () def _handle_event( event: cygrpc.BaseEvent, state: _RPCState, response_deserializer: Optional[DeserializingFunction], ) -> List[NullaryCallbackType]: callbacks = [] for batch_operation in event.batch_operations: operation_type = batch_operation.type() state.due.remove(operation_type) if operation_type == cygrpc.OperationType.receive_initial_metadata: state.initial_metadata = batch_operation.initial_metadata() elif operation_type == cygrpc.OperationType.receive_message: serialized_response = batch_operation.message() if serialized_response is not None: response = _common.deserialize( serialized_response, response_deserializer ) if response is None: details = "Exception deserializing response!" _abort(state, grpc.StatusCode.INTERNAL, details) else: state.response = response elif operation_type == cygrpc.OperationType.receive_status_on_client: state.trailing_metadata = batch_operation.trailing_metadata() if state.code is None: code = _common.CYGRPC_STATUS_CODE_TO_STATUS_CODE.get( batch_operation.code() ) if code is None: state.code = grpc.StatusCode.UNKNOWN state.details = _unknown_code_details( code, batch_operation.details() ) else: state.code = code state.details = batch_operation.details() state.debug_error_string = batch_operation.error_string() state.rpc_end_time = time.perf_counter() _observability.maybe_record_rpc_latency(state) callbacks.extend(state.callbacks) state.callbacks = None return callbacks def _event_handler( state: _RPCState, response_deserializer: Optional[DeserializingFunction] ) -> UserTag: def handle_event(event): with state.condition: callbacks = _handle_event(event, state, response_deserializer) state.condition.notify_all() done = not state.due for callback in callbacks: try: callback() except Exception as e: # pylint: disable=broad-except # NOTE(rbellevi): We suppress but log errors here so as not to # kill the channel spin thread. logging.error( "Exception in callback %s: %s", repr(callback.func), repr(e) ) return done and state.fork_epoch >= cygrpc.get_fork_epoch() return handle_event # TODO(xuanwn): Create a base class for IntegratedCall and SegregatedCall. # pylint: disable=too-many-statements def _consume_request_iterator( request_iterator: Iterator, state: _RPCState, call: Union[cygrpc.IntegratedCall, cygrpc.SegregatedCall], request_serializer: SerializingFunction, event_handler: Optional[UserTag], ) -> None: """Consume a request supplied by the user.""" def consume_request_iterator(): # pylint: disable=too-many-branches # Iterate over the request iterator until it is exhausted or an error # condition is encountered. while True: return_from_user_request_generator_invoked = False try: # The thread may die in user-code. Do not block fork for this. cygrpc.enter_user_request_generator() request = next(request_iterator) except StopIteration: break except Exception: # pylint: disable=broad-except cygrpc.return_from_user_request_generator() return_from_user_request_generator_invoked = True code = grpc.StatusCode.UNKNOWN details = "Exception iterating requests!" _LOGGER.exception(details) call.cancel( _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE[code], details ) _abort(state, code, details) return finally: if not return_from_user_request_generator_invoked: cygrpc.return_from_user_request_generator() serialized_request = _common.serialize(request, request_serializer) with state.condition: if state.code is None and not state.cancelled: if serialized_request is None: code = grpc.StatusCode.INTERNAL details = "Exception serializing request!" call.cancel( _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE[code], details, ) _abort(state, code, details) return else: state.due.add(cygrpc.OperationType.send_message) operations = ( cygrpc.SendMessageOperation( serialized_request, _EMPTY_FLAGS ), ) operating = call.operate(operations, event_handler) if not operating: state.due.remove(cygrpc.OperationType.send_message) return def _done(): return ( state.code is not None or cygrpc.OperationType.send_message not in state.due ) _common.wait( state.condition.wait, _done, spin_cb=functools.partial( cygrpc.block_if_fork_in_progress, state ), ) if state.code is not None: return else: return with state.condition: if state.code is None: state.due.add(cygrpc.OperationType.send_close_from_client) operations = ( cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS), ) operating = call.operate(operations, event_handler) if not operating: state.due.remove( cygrpc.OperationType.send_close_from_client ) consumption_thread = cygrpc.ForkManagedThread( target=consume_request_iterator ) consumption_thread.setDaemon(True) consumption_thread.start() def _rpc_state_string(class_name: str, rpc_state: _RPCState) -> str: """Calculates error string for RPC.""" with rpc_state.condition: if rpc_state.code is None: return "<{} object>".format(class_name) elif rpc_state.code is grpc.StatusCode.OK: return _OK_RENDEZVOUS_REPR_FORMAT.format( class_name, rpc_state.code, rpc_state.details ) else: return _NON_OK_RENDEZVOUS_REPR_FORMAT.format( class_name, rpc_state.code, rpc_state.details, rpc_state.debug_error_string, ) class _InactiveRpcError(grpc.RpcError, grpc.Call, grpc.Future): """An RPC error not tied to the execution of a particular RPC. The RPC represented by the state object must not be in-progress or cancelled. Attributes: _state: An instance of _RPCState. """ _state: _RPCState def __init__(self, state: _RPCState): with state.condition: self._state = _RPCState( (), copy.deepcopy(state.initial_metadata), copy.deepcopy(state.trailing_metadata), state.code, copy.deepcopy(state.details), ) self._state.response = copy.copy(state.response) self._state.debug_error_string = copy.copy(state.debug_error_string) def initial_metadata(self) -> Optional[MetadataType]: return self._state.initial_metadata def trailing_metadata(self) -> Optional[MetadataType]: return self._state.trailing_metadata def code(self) -> Optional[grpc.StatusCode]: return self._state.code def details(self) -> Optional[str]: return _common.decode(self._state.details) def debug_error_string(self) -> Optional[str]: return _common.decode(self._state.debug_error_string) def _repr(self) -> str: return _rpc_state_string(self.__class__.__name__, self._state) def __repr__(self) -> str: return self._repr() def __str__(self) -> str: return self._repr() def cancel(self) -> bool: """See grpc.Future.cancel.""" return False def cancelled(self) -> bool: """See grpc.Future.cancelled.""" return False def running(self) -> bool: """See grpc.Future.running.""" return False def done(self) -> bool: """See grpc.Future.done.""" return True def result( self, timeout: Optional[float] = None ) -> Any: # pylint: disable=unused-argument """See grpc.Future.result.""" raise self def exception( self, timeout: Optional[float] = None # pylint: disable=unused-argument ) -> Optional[Exception]: """See grpc.Future.exception.""" return self def traceback( self, timeout: Optional[float] = None # pylint: disable=unused-argument ) -> Optional[types.TracebackType]: """See grpc.Future.traceback.""" try: raise self except grpc.RpcError: return sys.exc_info()[2] def add_done_callback( self, fn: Callable[[grpc.Future], None], timeout: Optional[float] = None, # pylint: disable=unused-argument ) -> None: """See grpc.Future.add_done_callback.""" fn(self) class _Rendezvous(grpc.RpcError, grpc.RpcContext): """An RPC iterator. Attributes: _state: An instance of _RPCState. _call: An instance of SegregatedCall or IntegratedCall. In either case, the _call object is expected to have operate, cancel, and next_event methods. _response_deserializer: A callable taking bytes and return a Python object. _deadline: A float representing the deadline of the RPC in seconds. Or possibly None, to represent an RPC with no deadline at all. """ _state: _RPCState _call: Union[cygrpc.SegregatedCall, cygrpc.IntegratedCall] _response_deserializer: Optional[DeserializingFunction] _deadline: Optional[float] def __init__( self, state: _RPCState, call: Union[cygrpc.SegregatedCall, cygrpc.IntegratedCall], response_deserializer: Optional[DeserializingFunction], deadline: Optional[float], ): super(_Rendezvous, self).__init__() self._state = state self._call = call self._response_deserializer = response_deserializer self._deadline = deadline def is_active(self) -> bool: """See grpc.RpcContext.is_active""" with self._state.condition: return self._state.code is None def time_remaining(self) -> Optional[float]: """See grpc.RpcContext.time_remaining""" with self._state.condition: if self._deadline is None: return None else: return max(self._deadline - time.time(), 0) def cancel(self) -> bool: """See grpc.RpcContext.cancel""" with self._state.condition: if self._state.code is None: code = grpc.StatusCode.CANCELLED details = "Locally cancelled by application!" self._call.cancel( _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE[code], details ) self._state.cancelled = True _abort(self._state, code, details) self._state.condition.notify_all() return True else: return False def add_callback(self, callback: NullaryCallbackType) -> bool: """See grpc.RpcContext.add_callback""" with self._state.condition: if self._state.callbacks is None: return False else: self._state.callbacks.append(callback) return True def __iter__(self): return self def next(self): return self._next() def __next__(self): return self._next() def _next(self): raise NotImplementedError() def debug_error_string(self) -> Optional[str]: raise NotImplementedError() def _repr(self) -> str: return _rpc_state_string(self.__class__.__name__, self._state) def __repr__(self) -> str: return self._repr() def __str__(self) -> str: return self._repr() def __del__(self) -> None: with self._state.condition: if self._state.code is None: self._state.code = grpc.StatusCode.CANCELLED self._state.details = "Cancelled upon garbage collection!" self._state.cancelled = True self._call.cancel( _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE[self._state.code], self._state.details, ) self._state.condition.notify_all() class _SingleThreadedRendezvous( _Rendezvous, grpc.Call, grpc.Future ): # pylint: disable=too-many-ancestors """An RPC iterator operating entirely on a single thread. The __next__ method of _SingleThreadedRendezvous does not depend on the existence of any other thread, including the "channel spin thread". However, this means that its interface is entirely synchronous. So this class cannot completely fulfill the grpc.Future interface. The result, exception, and traceback methods will never block and will instead raise an exception if calling the method would result in blocking. This means that these methods are safe to call from add_done_callback handlers. """ _state: _RPCState def _is_complete(self) -> bool: return self._state.code is not None def cancelled(self) -> bool: with self._state.condition: return self._state.cancelled def running(self) -> bool: with self._state.condition: return self._state.code is None def done(self) -> bool: with self._state.condition: return self._state.code is not None def result(self, timeout: Optional[float] = None) -> Any: """Returns the result of the computation or raises its exception. This method will never block. Instead, it will raise an exception if calling this method would otherwise result in blocking. Since this method will never block, any `timeout` argument passed will be ignored. """ del timeout with self._state.condition: if not self._is_complete(): raise grpc.experimental.UsageError( "_SingleThreadedRendezvous only supports result() when the" " RPC is complete." ) if self._state.code is grpc.StatusCode.OK: return self._state.response elif self._state.cancelled: raise grpc.FutureCancelledError() else: raise self def exception(self, timeout: Optional[float] = None) -> Optional[Exception]: """Return the exception raised by the computation. This method will never block. Instead, it will raise an exception if calling this method would otherwise result in blocking. Since this method will never block, any `timeout` argument passed will be ignored. """ del timeout with self._state.condition: if not self._is_complete(): raise grpc.experimental.UsageError( "_SingleThreadedRendezvous only supports exception() when" " the RPC is complete." ) if self._state.code is grpc.StatusCode.OK: return None elif self._state.cancelled: raise grpc.FutureCancelledError() else: return self def traceback( self, timeout: Optional[float] = None ) -> Optional[types.TracebackType]: """Access the traceback of the exception raised by the computation. This method will never block. Instead, it will raise an exception if calling this method would otherwise result in blocking. Since this method will never block, any `timeout` argument passed will be ignored. """ del timeout with self._state.condition: if not self._is_complete(): raise grpc.experimental.UsageError( "_SingleThreadedRendezvous only supports traceback() when" " the RPC is complete." ) if self._state.code is grpc.StatusCode.OK: return None elif self._state.cancelled: raise grpc.FutureCancelledError() else: try: raise self except grpc.RpcError: return sys.exc_info()[2] def add_done_callback(self, fn: Callable[[grpc.Future], None]) -> None: with self._state.condition: if self._state.code is None: self._state.callbacks.append(functools.partial(fn, self)) return fn(self) def initial_metadata(self) -> Optional[MetadataType]: """See grpc.Call.initial_metadata""" with self._state.condition: # NOTE(gnossen): Based on our initial call batch, we are guaranteed # to receive initial metadata before any messages. while self._state.initial_metadata is None: self._consume_next_event() return self._state.initial_metadata def trailing_metadata(self) -> Optional[MetadataType]: """See grpc.Call.trailing_metadata""" with self._state.condition: if self._state.trailing_metadata is None: raise grpc.experimental.UsageError( "Cannot get trailing metadata until RPC is completed." ) return self._state.trailing_metadata def code(self) -> Optional[grpc.StatusCode]: """See grpc.Call.code""" with self._state.condition: if self._state.code is None: raise grpc.experimental.UsageError( "Cannot get code until RPC is completed." ) return self._state.code def details(self) -> Optional[str]: """See grpc.Call.details""" with self._state.condition: if self._state.details is None: raise grpc.experimental.UsageError( "Cannot get details until RPC is completed." ) return _common.decode(self._state.details) def _consume_next_event(self) -> Optional[cygrpc.BaseEvent]: event = self._call.next_event() with self._state.condition: callbacks = _handle_event( event, self._state, self._response_deserializer ) for callback in callbacks: # NOTE(gnossen): We intentionally allow exceptions to bubble up # to the user when running on a single thread. callback() return event def _next_response(self) -> Any: while True: self._consume_next_event() with self._state.condition: if self._state.response is not None: response = self._state.response self._state.response = None return response elif ( cygrpc.OperationType.receive_message not in self._state.due ): if self._state.code is grpc.StatusCode.OK: raise StopIteration() elif self._state.code is not None: raise self def _next(self) -> Any: with self._state.condition: if self._state.code is None: # We tentatively add the operation as expected and remove # it if the enqueue operation fails. This allows us to guarantee that # if an event has been submitted to the core completion queue, # it is in `due`. If we waited until after a successful # enqueue operation then a signal could interrupt this # thread between the enqueue operation and the addition of the # operation to `due`. This would cause an exception on the # channel spin thread when the operation completes and no # corresponding operation would be present in state.due. # Note that, since `condition` is held through this block, there is # no data race on `due`. self._state.due.add(cygrpc.OperationType.receive_message) operating = self._call.operate( (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),), None ) if not operating: self._state.due.remove(cygrpc.OperationType.receive_message) elif self._state.code is grpc.StatusCode.OK: raise StopIteration() else: raise self return self._next_response() def debug_error_string(self) -> Optional[str]: with self._state.condition: if self._state.debug_error_string is None: raise grpc.experimental.UsageError( "Cannot get debug error string until RPC is completed." ) return _common.decode(self._state.debug_error_string) class _MultiThreadedRendezvous( _Rendezvous, grpc.Call, grpc.Future ): # pylint: disable=too-many-ancestors """An RPC iterator that depends on a channel spin thread. This iterator relies upon a per-channel thread running in the background, dequeueing events from the completion queue, and notifying threads waiting on the threading.Condition object in the _RPCState object. This extra thread allows _MultiThreadedRendezvous to fulfill the grpc.Future interface and to mediate a bidirection streaming RPC. """ _state: _RPCState def initial_metadata(self) -> Optional[MetadataType]: """See grpc.Call.initial_metadata""" with self._state.condition: def _done(): return self._state.initial_metadata is not None _common.wait(self._state.condition.wait, _done) return self._state.initial_metadata def trailing_metadata(self) -> Optional[MetadataType]: """See grpc.Call.trailing_metadata""" with self._state.condition: def _done(): return self._state.trailing_metadata is not None _common.wait(self._state.condition.wait, _done) return self._state.trailing_metadata def code(self) -> Optional[grpc.StatusCode]: """See grpc.Call.code""" with self._state.condition: def _done(): return self._state.code is not None _common.wait(self._state.condition.wait, _done) return self._state.code def details(self) -> Optional[str]: """See grpc.Call.details""" with self._state.condition: def _done(): return self._state.details is not None _common.wait(self._state.condition.wait, _done) return _common.decode(self._state.details) def debug_error_string(self) -> Optional[str]: with self._state.condition: def _done(): return self._state.debug_error_string is not None _common.wait(self._state.condition.wait, _done) return _common.decode(self._state.debug_error_string) def cancelled(self) -> bool: with self._state.condition: return self._state.cancelled def running(self) -> bool: with self._state.condition: return self._state.code is None def done(self) -> bool: with self._state.condition: return self._state.code is not None def _is_complete(self) -> bool: return self._state.code is not None def result(self, timeout: Optional[float] = None) -> Any: """Returns the result of the computation or raises its exception. See grpc.Future.result for the full API contract. """ with self._state.condition: timed_out = _common.wait( self._state.condition.wait, self._is_complete, timeout=timeout ) if timed_out: raise grpc.FutureTimeoutError() else: if self._state.code is grpc.StatusCode.OK: return self._state.response elif self._state.cancelled: raise grpc.FutureCancelledError() else: raise self def exception(self, timeout: Optional[float] = None) -> Optional[Exception]: """Return the exception raised by the computation. See grpc.Future.exception for the full API contract. """ with self._state.condition: timed_out = _common.wait( self._state.condition.wait, self._is_complete, timeout=timeout ) if timed_out: raise grpc.FutureTimeoutError() else: if self._state.code is grpc.StatusCode.OK: return None elif self._state.cancelled: raise grpc.FutureCancelledError() else: return self def traceback( self, timeout: Optional[float] = None ) -> Optional[types.TracebackType]: """Access the traceback of the exception raised by the computation. See grpc.future.traceback for the full API contract. """ with self._state.condition: timed_out = _common.wait( self._state.condition.wait, self._is_complete, timeout=timeout ) if timed_out: raise grpc.FutureTimeoutError() else: if self._state.code is grpc.StatusCode.OK: return None elif self._state.cancelled: raise grpc.FutureCancelledError() else: try: raise self except grpc.RpcError: return sys.exc_info()[2] def add_done_callback(self, fn: Callable[[grpc.Future], None]) -> None: with self._state.condition: if self._state.code is None: self._state.callbacks.append(functools.partial(fn, self)) return fn(self) def _next(self) -> Any: with self._state.condition: if self._state.code is None: event_handler = _event_handler( self._state, self._response_deserializer ) self._state.due.add(cygrpc.OperationType.receive_message) operating = self._call.operate( (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),), event_handler, ) if not operating: self._state.due.remove(cygrpc.OperationType.receive_message) elif self._state.code is grpc.StatusCode.OK: raise StopIteration() else: raise self def _response_ready(): return self._state.response is not None or ( cygrpc.OperationType.receive_message not in self._state.due and self._state.code is not None ) _common.wait(self._state.condition.wait, _response_ready) if self._state.response is not None: response = self._state.response self._state.response = None return response elif cygrpc.OperationType.receive_message not in self._state.due: if self._state.code is grpc.StatusCode.OK: raise StopIteration() elif self._state.code is not None: raise self def _start_unary_request( request: Any, timeout: Optional[float], request_serializer: SerializingFunction, ) -> Tuple[Optional[float], Optional[bytes], Optional[grpc.RpcError]]: deadline = _deadline(timeout) serialized_request = _common.serialize(request, request_serializer) if serialized_request is None: state = _RPCState( (), (), (), grpc.StatusCode.INTERNAL, "Exception serializing request!", ) error = _InactiveRpcError(state) return deadline, None, error else: return deadline, serialized_request, None def _end_unary_response_blocking( state: _RPCState, call: cygrpc.SegregatedCall, with_call: bool, deadline: Optional[float], ) -> Union[ResponseType, Tuple[ResponseType, grpc.Call]]: if state.code is grpc.StatusCode.OK: if with_call: rendezvous = _MultiThreadedRendezvous(state, call, None, deadline) return state.response, rendezvous else: return state.response else: raise _InactiveRpcError(state) # pytype: disable=not-instantiable def _stream_unary_invocation_operations( metadata: Optional[MetadataType], initial_metadata_flags: int ) -> Sequence[Sequence[cygrpc.Operation]]: return ( ( cygrpc.SendInitialMetadataOperation( metadata, initial_metadata_flags ), cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS), cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS), ), (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),), ) def _stream_unary_invocation_operations_and_tags( metadata: Optional[MetadataType], initial_metadata_flags: int ) -> Sequence[Tuple[Sequence[cygrpc.Operation], Optional[UserTag]]]: return tuple( ( operations, None, ) for operations in _stream_unary_invocation_operations( metadata, initial_metadata_flags ) ) def _determine_deadline(user_deadline: Optional[float]) -> Optional[float]: parent_deadline = cygrpc.get_deadline_from_context() if parent_deadline is None and user_deadline is None: return None elif parent_deadline is not None and user_deadline is None: return parent_deadline elif user_deadline is not None and parent_deadline is None: return user_deadline else: return min(parent_deadline, user_deadline) class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): _channel: cygrpc.Channel _managed_call: IntegratedCallFactory _method: bytes _target: bytes _request_serializer: Optional[SerializingFunction] _response_deserializer: Optional[DeserializingFunction] _context: Any _registered_call_handle: Optional[int] __slots__ = [ "_channel", "_managed_call", "_method", "_target", "_request_serializer", "_response_deserializer", "_context", ] # pylint: disable=too-many-arguments def __init__( self, channel: cygrpc.Channel, managed_call: IntegratedCallFactory, method: bytes, target: bytes, request_serializer: Optional[SerializingFunction], response_deserializer: Optional[DeserializingFunction], _registered_call_handle: Optional[int], ): self._channel = channel self._managed_call = managed_call self._method = method self._target = target self._request_serializer = request_serializer self._response_deserializer = response_deserializer self._context = cygrpc.build_census_context() self._registered_call_handle = _registered_call_handle def _prepare( self, request: Any, timeout: Optional[float], metadata: Optional[MetadataType], wait_for_ready: Optional[bool], compression: Optional[grpc.Compression], ) -> Tuple[ Optional[_RPCState], Optional[Sequence[cygrpc.Operation]], Optional[float], Optional[grpc.RpcError], ]: deadline, serialized_request, rendezvous = _start_unary_request( request, timeout, self._request_serializer ) initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( wait_for_ready ) augmented_metadata = _compression.augment_metadata( metadata, compression ) if serialized_request is None: return None, None, None, rendezvous else: state = _RPCState(_UNARY_UNARY_INITIAL_DUE, None, None, None, None) operations = ( cygrpc.SendInitialMetadataOperation( augmented_metadata, initial_metadata_flags ), cygrpc.SendMessageOperation(serialized_request, _EMPTY_FLAGS), cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS), cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS), cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS), cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS), ) return state, operations, deadline, None def _blocking( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Tuple[_RPCState, cygrpc.SegregatedCall]: state, operations, deadline, rendezvous = self._prepare( request, timeout, metadata, wait_for_ready, compression ) if state is None: raise rendezvous # pylint: disable-msg=raising-bad-type else: state.rpc_start_time = time.perf_counter() state.method = _common.decode(self._method) state.target = _common.decode(self._target) call = self._channel.segregated_call( cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS, self._method, None, _determine_deadline(deadline), metadata, None if credentials is None else credentials._credentials, ( ( operations, None, ), ), self._context, self._registered_call_handle, ) event = call.next_event() _handle_event(event, state, self._response_deserializer) return state, call def __call__( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Any: ( state, call, ) = self._blocking( request, timeout, metadata, credentials, wait_for_ready, compression ) return _end_unary_response_blocking(state, call, False, None) def with_call( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Tuple[Any, grpc.Call]: ( state, call, ) = self._blocking( request, timeout, metadata, credentials, wait_for_ready, compression ) return _end_unary_response_blocking(state, call, True, None) def future( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> _MultiThreadedRendezvous: state, operations, deadline, rendezvous = self._prepare( request, timeout, metadata, wait_for_ready, compression ) if state is None: raise rendezvous # pylint: disable-msg=raising-bad-type else: event_handler = _event_handler(state, self._response_deserializer) state.rpc_start_time = time.perf_counter() state.method = _common.decode(self._method) state.target = _common.decode(self._target) call = self._managed_call( cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS, self._method, None, deadline, metadata, None if credentials is None else credentials._credentials, (operations,), event_handler, self._context, self._registered_call_handle, ) return _MultiThreadedRendezvous( state, call, self._response_deserializer, deadline ) class _SingleThreadedUnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): _channel: cygrpc.Channel _method: bytes _target: bytes _request_serializer: Optional[SerializingFunction] _response_deserializer: Optional[DeserializingFunction] _context: Any _registered_call_handle: Optional[int] __slots__ = [ "_channel", "_method", "_target", "_request_serializer", "_response_deserializer", "_context", ] # pylint: disable=too-many-arguments def __init__( self, channel: cygrpc.Channel, method: bytes, target: bytes, request_serializer: SerializingFunction, response_deserializer: DeserializingFunction, _registered_call_handle: Optional[int], ): self._channel = channel self._method = method self._target = target self._request_serializer = request_serializer self._response_deserializer = response_deserializer self._context = cygrpc.build_census_context() self._registered_call_handle = _registered_call_handle def __call__( # pylint: disable=too-many-locals self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> _SingleThreadedRendezvous: deadline = _deadline(timeout) serialized_request = _common.serialize( request, self._request_serializer ) if serialized_request is None: state = _RPCState( (), (), (), grpc.StatusCode.INTERNAL, "Exception serializing request!", ) raise _InactiveRpcError(state) state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None) call_credentials = ( None if credentials is None else credentials._credentials ) initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( wait_for_ready ) augmented_metadata = _compression.augment_metadata( metadata, compression ) operations = ( ( cygrpc.SendInitialMetadataOperation( augmented_metadata, initial_metadata_flags ), cygrpc.SendMessageOperation(serialized_request, _EMPTY_FLAGS), cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS), ), (cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS),), (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),), ) operations_and_tags = tuple((ops, None) for ops in operations) state.rpc_start_time = time.perf_counter() state.method = _common.decode(self._method) state.target = _common.decode(self._target) call = self._channel.segregated_call( cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS, self._method, None, _determine_deadline(deadline), metadata, call_credentials, operations_and_tags, self._context, self._registered_call_handle, ) return _SingleThreadedRendezvous( state, call, self._response_deserializer, deadline ) class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): _channel: cygrpc.Channel _managed_call: IntegratedCallFactory _method: bytes _target: bytes _request_serializer: Optional[SerializingFunction] _response_deserializer: Optional[DeserializingFunction] _context: Any _registered_call_handle: Optional[int] __slots__ = [ "_channel", "_managed_call", "_method", "_target", "_request_serializer", "_response_deserializer", "_context", ] # pylint: disable=too-many-arguments def __init__( self, channel: cygrpc.Channel, managed_call: IntegratedCallFactory, method: bytes, target: bytes, request_serializer: SerializingFunction, response_deserializer: DeserializingFunction, _registered_call_handle: Optional[int], ): self._channel = channel self._managed_call = managed_call self._method = method self._target = target self._request_serializer = request_serializer self._response_deserializer = response_deserializer self._context = cygrpc.build_census_context() self._registered_call_handle = _registered_call_handle def __call__( # pylint: disable=too-many-locals self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> _MultiThreadedRendezvous: deadline, serialized_request, rendezvous = _start_unary_request( request, timeout, self._request_serializer ) initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( wait_for_ready ) if serialized_request is None: raise rendezvous # pylint: disable-msg=raising-bad-type else: augmented_metadata = _compression.augment_metadata( metadata, compression ) state = _RPCState(_UNARY_STREAM_INITIAL_DUE, None, None, None, None) operations = ( ( cygrpc.SendInitialMetadataOperation( augmented_metadata, initial_metadata_flags ), cygrpc.SendMessageOperation( serialized_request, _EMPTY_FLAGS ), cygrpc.SendCloseFromClientOperation(_EMPTY_FLAGS), cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS), ), (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),), ) state.rpc_start_time = time.perf_counter() state.method = _common.decode(self._method) state.target = _common.decode(self._target) call = self._managed_call( cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS, self._method, None, _determine_deadline(deadline), metadata, None if credentials is None else credentials._credentials, operations, _event_handler(state, self._response_deserializer), self._context, self._registered_call_handle, ) return _MultiThreadedRendezvous( state, call, self._response_deserializer, deadline ) class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): _channel: cygrpc.Channel _managed_call: IntegratedCallFactory _method: bytes _target: bytes _request_serializer: Optional[SerializingFunction] _response_deserializer: Optional[DeserializingFunction] _context: Any _registered_call_handle: Optional[int] __slots__ = [ "_channel", "_managed_call", "_method", "_target", "_request_serializer", "_response_deserializer", "_context", ] # pylint: disable=too-many-arguments def __init__( self, channel: cygrpc.Channel, managed_call: IntegratedCallFactory, method: bytes, target: bytes, request_serializer: Optional[SerializingFunction], response_deserializer: Optional[DeserializingFunction], _registered_call_handle: Optional[int], ): self._channel = channel self._managed_call = managed_call self._method = method self._target = target self._request_serializer = request_serializer self._response_deserializer = response_deserializer self._context = cygrpc.build_census_context() self._registered_call_handle = _registered_call_handle def _blocking( self, request_iterator: Iterator, timeout: Optional[float], metadata: Optional[MetadataType], credentials: Optional[grpc.CallCredentials], wait_for_ready: Optional[bool], compression: Optional[grpc.Compression], ) -> Tuple[_RPCState, cygrpc.SegregatedCall]: deadline = _deadline(timeout) state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( wait_for_ready ) augmented_metadata = _compression.augment_metadata( metadata, compression ) state.rpc_start_time = time.perf_counter() state.method = _common.decode(self._method) state.target = _common.decode(self._target) call = self._channel.segregated_call( cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS, self._method, None, _determine_deadline(deadline), augmented_metadata, None if credentials is None else credentials._credentials, _stream_unary_invocation_operations_and_tags( augmented_metadata, initial_metadata_flags ), self._context, self._registered_call_handle, ) _consume_request_iterator( request_iterator, state, call, self._request_serializer, None ) while True: event = call.next_event() with state.condition: _handle_event(event, state, self._response_deserializer) state.condition.notify_all() if not state.due: break return state, call def __call__( self, request_iterator: Iterator, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Any: ( state, call, ) = self._blocking( request_iterator, timeout, metadata, credentials, wait_for_ready, compression, ) return _end_unary_response_blocking(state, call, False, None) def with_call( self, request_iterator: Iterator, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Tuple[Any, grpc.Call]: ( state, call, ) = self._blocking( request_iterator, timeout, metadata, credentials, wait_for_ready, compression, ) return _end_unary_response_blocking(state, call, True, None) def future( self, request_iterator: Iterator, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> _MultiThreadedRendezvous: deadline = _deadline(timeout) state = _RPCState(_STREAM_UNARY_INITIAL_DUE, None, None, None, None) event_handler = _event_handler(state, self._response_deserializer) initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( wait_for_ready ) augmented_metadata = _compression.augment_metadata( metadata, compression ) state.rpc_start_time = time.perf_counter() state.method = _common.decode(self._method) state.target = _common.decode(self._target) call = self._managed_call( cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS, self._method, None, deadline, augmented_metadata, None if credentials is None else credentials._credentials, _stream_unary_invocation_operations( metadata, initial_metadata_flags ), event_handler, self._context, self._registered_call_handle, ) _consume_request_iterator( request_iterator, state, call, self._request_serializer, event_handler, ) return _MultiThreadedRendezvous( state, call, self._response_deserializer, deadline ) class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): _channel: cygrpc.Channel _managed_call: IntegratedCallFactory _method: bytes _target: bytes _request_serializer: Optional[SerializingFunction] _response_deserializer: Optional[DeserializingFunction] _context: Any _registered_call_handle: Optional[int] __slots__ = [ "_channel", "_managed_call", "_method", "_target", "_request_serializer", "_response_deserializer", "_context", ] # pylint: disable=too-many-arguments def __init__( self, channel: cygrpc.Channel, managed_call: IntegratedCallFactory, method: bytes, target: bytes, request_serializer: Optional[SerializingFunction], response_deserializer: Optional[DeserializingFunction], _registered_call_handle: Optional[int], ): self._channel = channel self._managed_call = managed_call self._method = method self._target = target self._request_serializer = request_serializer self._response_deserializer = response_deserializer self._context = cygrpc.build_census_context() self._registered_call_handle = _registered_call_handle def __call__( self, request_iterator: Iterator, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> _MultiThreadedRendezvous: deadline = _deadline(timeout) state = _RPCState(_STREAM_STREAM_INITIAL_DUE, None, None, None, None) initial_metadata_flags = _InitialMetadataFlags().with_wait_for_ready( wait_for_ready ) augmented_metadata = _compression.augment_metadata( metadata, compression ) operations = ( ( cygrpc.SendInitialMetadataOperation( augmented_metadata, initial_metadata_flags ), cygrpc.ReceiveStatusOnClientOperation(_EMPTY_FLAGS), ), (cygrpc.ReceiveInitialMetadataOperation(_EMPTY_FLAGS),), ) event_handler = _event_handler(state, self._response_deserializer) state.rpc_start_time = time.perf_counter() state.method = _common.decode(self._method) state.target = _common.decode(self._target) call = self._managed_call( cygrpc.PropagationConstants.GRPC_PROPAGATE_DEFAULTS, self._method, None, _determine_deadline(deadline), augmented_metadata, None if credentials is None else credentials._credentials, operations, event_handler, self._context, self._registered_call_handle, ) _consume_request_iterator( request_iterator, state, call, self._request_serializer, event_handler, ) return _MultiThreadedRendezvous( state, call, self._response_deserializer, deadline ) class _InitialMetadataFlags(int): """Stores immutable initial metadata flags""" def __new__(cls, value: int = _EMPTY_FLAGS): value &= cygrpc.InitialMetadataFlags.used_mask return super(_InitialMetadataFlags, cls).__new__(cls, value) def with_wait_for_ready(self, wait_for_ready: Optional[bool]) -> int: if wait_for_ready is not None: if wait_for_ready: return self.__class__( self \| cygrpc.InitialMetadataFlags.wait_for_ready \| cygrpc.InitialMetadataFlags.wait_for_ready_explicitly_set ) elif not wait_for_ready: return self.__class__( self & ~cygrpc.InitialMetadataFlags.wait_for_ready \| cygrpc.InitialMetadataFlags.wait_for_ready_explicitly_set ) return self class _ChannelCallState(object): channel: cygrpc.Channel managed_calls: int threading: bool def __init__(self, channel: cygrpc.Channel): self.lock = threading.Lock() self.channel = channel self.managed_calls = 0 self.threading = False def reset_postfork_child(self) -> None: self.managed_calls = 0 def __del__(self): try: self.channel.close( cygrpc.StatusCode.cancelled, "Channel deallocated!" ) except (TypeError, AttributeError): pass def _run_channel_spin_thread(state: _ChannelCallState) -> None: def channel_spin(): while True: cygrpc.block_if_fork_in_progress(state) event = state.channel.next_call_event() if event.completion_type == cygrpc.CompletionType.queue_timeout: continue call_completed = event.tag(event) if call_completed: with state.lock: state.managed_calls -= 1 if state.managed_calls == 0: return channel_spin_thread = cygrpc.ForkManagedThread(target=channel_spin) channel_spin_thread.setDaemon(True) channel_spin_thread.start() def _channel_managed_call_management(state: _ChannelCallState): # pylint: disable=too-many-arguments def create( flags: int, method: bytes, host: Optional[str], deadline: Optional[float], metadata: Optional[MetadataType], credentials: Optional[cygrpc.CallCredentials], operations: Sequence[Sequence[cygrpc.Operation]], event_handler: UserTag, context: Any, _registered_call_handle: Optional[int], ) -> cygrpc.IntegratedCall: """Creates a cygrpc.IntegratedCall. Args: flags: An integer bitfield of call flags. method: The RPC method. host: A host string for the created call. deadline: A float to be the deadline of the created call or None if the call is to have an infinite deadline. metadata: The metadata for the call or None. credentials: A cygrpc.CallCredentials or None. operations: A sequence of sequences of cygrpc.Operations to be started on the call. event_handler: A behavior to call to handle the events resultant from the operations on the call. context: Context object for distributed tracing. _registered_call_handle: An int representing the call handle of the method, or None if the method is not registered. Returns: A cygrpc.IntegratedCall with which to conduct an RPC. """ operations_and_tags = tuple( ( operation, event_handler, ) for operation in operations ) with state.lock: call = state.channel.integrated_call( flags, method, host, deadline, metadata, credentials, operations_and_tags, context, _registered_call_handle, ) if state.managed_calls == 0: state.managed_calls = 1 _run_channel_spin_thread(state) else: state.managed_calls += 1 return call return create class _ChannelConnectivityState(object): lock: threading.RLock channel: grpc.Channel polling: bool connectivity: grpc.ChannelConnectivity try_to_connect: bool # TODO(xuanwn): Refactor this: https://github.com/grpc/grpc/issues/31704 callbacks_and_connectivities: List[ Sequence[ Union[ Callable[[grpc.ChannelConnectivity], None], Optional[grpc.ChannelConnectivity], ] ] ] delivering: bool def __init__(self, channel: grpc.Channel): self.lock = threading.RLock() self.channel = channel self.polling = False self.connectivity = None self.try_to_connect = False self.callbacks_and_connectivities = [] self.delivering = False def reset_postfork_child(self) -> None: self.polling = False self.connectivity = None self.try_to_connect = False self.callbacks_and_connectivities = [] self.delivering = False def _deliveries( state: _ChannelConnectivityState, ) -> List[Callable[[grpc.ChannelConnectivity], None]]: callbacks_needing_update = [] for callback_and_connectivity in state.callbacks_and_connectivities: ( callback, callback_connectivity, ) = callback_and_connectivity if callback_connectivity is not state.connectivity: callbacks_needing_update.append(callback) callback_and_connectivity[1] = state.connectivity return callbacks_needing_update def _deliver( state: _ChannelConnectivityState, initial_connectivity: grpc.ChannelConnectivity, initial_callbacks: Sequence[Callable[[grpc.ChannelConnectivity], None]], ) -> None: connectivity = initial_connectivity callbacks = initial_callbacks while True: for callback in callbacks: cygrpc.block_if_fork_in_progress(state) try: callback(connectivity) except Exception: # pylint: disable=broad-except _LOGGER.exception( _CHANNEL_SUBSCRIPTION_CALLBACK_ERROR_LOG_MESSAGE ) with state.lock: callbacks = _deliveries(state) if callbacks: connectivity = state.connectivity else: state.delivering = False return def _spawn_delivery( state: _ChannelConnectivityState, callbacks: Sequence[Callable[[grpc.ChannelConnectivity], None]], ) -> None: delivering_thread = cygrpc.ForkManagedThread( target=_deliver, args=( state, state.connectivity, callbacks, ), ) delivering_thread.setDaemon(True) delivering_thread.start() state.delivering = True # NOTE(https://github.com/grpc/grpc/issues/3064): We'd rather not poll. def _poll_connectivity( state: _ChannelConnectivityState, channel: grpc.Channel, initial_try_to_connect: bool, ) -> None: try_to_connect = initial_try_to_connect connectivity = channel.check_connectivity_state(try_to_connect) with state.lock: state.connectivity = ( _common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[ connectivity ] ) callbacks = tuple( callback for callback, _ in state.callbacks_and_connectivities ) for callback_and_connectivity in state.callbacks_and_connectivities: callback_and_connectivity[1] = state.connectivity if callbacks: _spawn_delivery(state, callbacks) while True: event = channel.watch_connectivity_state( connectivity, time.time() + 0.2 ) cygrpc.block_if_fork_in_progress(state) with state.lock: if ( not state.callbacks_and_connectivities and not state.try_to_connect ): state.polling = False state.connectivity = None break try_to_connect = state.try_to_connect state.try_to_connect = False if event.success or try_to_connect: connectivity = channel.check_connectivity_state(try_to_connect) with state.lock: state.connectivity = ( _common.CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY[ connectivity ] ) if not state.delivering: callbacks = _deliveries(state) if callbacks: _spawn_delivery(state, callbacks) def _subscribe( state: _ChannelConnectivityState, callback: Callable[[grpc.ChannelConnectivity], None], try_to_connect: bool, ) -> None: with state.lock: if not state.callbacks_and_connectivities and not state.polling: polling_thread = cygrpc.ForkManagedThread( target=_poll_connectivity, args=(state, state.channel, bool(try_to_connect)), ) polling_thread.setDaemon(True) polling_thread.start() state.polling = True state.callbacks_and_connectivities.append([callback, None]) elif not state.delivering and state.connectivity is not None: _spawn_delivery(state, (callback,)) state.try_to_connect \|= bool(try_to_connect) state.callbacks_and_connectivities.append( [callback, state.connectivity] ) else: state.try_to_connect \|= bool(try_to_connect) state.callbacks_and_connectivities.append([callback, None]) def _unsubscribe( state: _ChannelConnectivityState, callback: Callable[[grpc.ChannelConnectivity], None], ) -> None: with state.lock: for index, (subscribed_callback, unused_connectivity) in enumerate( state.callbacks_and_connectivities ): if callback == subscribed_callback: state.callbacks_and_connectivities.pop(index) break def _augment_options( base_options: Sequence[ChannelArgumentType], compression: Optional[grpc.Compression], ) -> Sequence[ChannelArgumentType]: compression_option = _compression.create_channel_option(compression) return ( tuple(base_options) + compression_option + ( ( cygrpc.ChannelArgKey.primary_user_agent_string, _USER_AGENT, ), ) ) def _separate_channel_options( options: Sequence[ChannelArgumentType], ) -> Tuple[Sequence[ChannelArgumentType], Sequence[ChannelArgumentType]]: """Separates core channel options from Python channel options.""" core_options = [] python_options = [] for pair in options: if ( pair[0] == grpc.experimental.ChannelOptions.SingleThreadedUnaryStream ): python_options.append(pair) else: core_options.append(pair) return python_options, core_options class Channel(grpc.Channel): """A cygrpc.Channel-backed implementation of grpc.Channel.""" _single_threaded_unary_stream: bool _channel: cygrpc.Channel _call_state: _ChannelCallState _connectivity_state: _ChannelConnectivityState _target: str _registered_call_handles: Dict[str, int] def __init__( self, target: str, options: Sequence[ChannelArgumentType], credentials: Optional[grpc.ChannelCredentials], compression: Optional[grpc.Compression], ): """Constructor. Args: target: The target to which to connect. options: Configuration options for the channel. credentials: A cygrpc.ChannelCredentials or None. compression: An optional value indicating the compression method to be used over the lifetime of the channel. """ python_options, core_options = _separate_channel_options(options) self._single_threaded_unary_stream = ( _DEFAULT_SINGLE_THREADED_UNARY_STREAM ) self._process_python_options(python_options) self._channel = cygrpc.Channel( _common.encode(target), _augment_options(core_options, compression), credentials, ) self._target = target self._call_state = _ChannelCallState(self._channel) self._connectivity_state = _ChannelConnectivityState(self._channel) cygrpc.fork_register_channel(self) if cygrpc.g_gevent_activated: cygrpc.gevent_increment_channel_count() def _get_registered_call_handle(self, method: str) -> int: """ Get the registered call handle for a method. This is a semi-private method. It is intended for use only by gRPC generated code. This method is not thread-safe. Args: method: Required, the method name for the RPC. Returns: The registered call handle pointer in the form of a Python Long. """ return self._channel.get_registered_call_handle(_common.encode(method)) def _process_python_options( self, python_options: Sequence[ChannelArgumentType] ) -> None: """Sets channel attributes according to python-only channel options.""" for pair in python_options: if ( pair[0] == grpc.experimental.ChannelOptions.SingleThreadedUnaryStream ): self._single_threaded_unary_stream = True def subscribe( self, callback: Callable[[grpc.ChannelConnectivity], None], try_to_connect: Optional[bool] = None, ) -> None: _subscribe(self._connectivity_state, callback, try_to_connect) def unsubscribe( self, callback: Callable[[grpc.ChannelConnectivity], None] ) -> None: _unsubscribe(self._connectivity_state, callback) # pylint: disable=arguments-differ def unary_unary( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None, _registered_method: Optional[bool] = False, ) -> grpc.UnaryUnaryMultiCallable: _registered_call_handle = None if _registered_method: _registered_call_handle = self._get_registered_call_handle(method) return _UnaryUnaryMultiCallable( self._channel, _channel_managed_call_management(self._call_state), _common.encode(method), _common.encode(self._target), request_serializer, response_deserializer, _registered_call_handle, ) # pylint: disable=arguments-differ def unary_stream( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None, _registered_method: Optional[bool] = False, ) -> grpc.UnaryStreamMultiCallable: _registered_call_handle = None if _registered_method: _registered_call_handle = self._get_registered_call_handle(method) # NOTE(rbellevi): Benchmarks have shown that running a unary-stream RPC # on a single Python thread results in an appreciable speed-up. However, # due to slight differences in capability, the multi-threaded variant # remains the default. if self._single_threaded_unary_stream: return _SingleThreadedUnaryStreamMultiCallable( self._channel, _common.encode(method), _common.encode(self._target), request_serializer, response_deserializer, _registered_call_handle, ) else: return _UnaryStreamMultiCallable( self._channel, _channel_managed_call_management(self._call_state), _common.encode(method), _common.encode(self._target), request_serializer, response_deserializer, _registered_call_handle, ) # pylint: disable=arguments-differ def stream_unary( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None, _registered_method: Optional[bool] = False, ) -> grpc.StreamUnaryMultiCallable: _registered_call_handle = None if _registered_method: _registered_call_handle = self._get_registered_call_handle(method) return _StreamUnaryMultiCallable( self._channel, _channel_managed_call_management(self._call_state), _common.encode(method), _common.encode(self._target), request_serializer, response_deserializer, _registered_call_handle, ) # pylint: disable=arguments-differ def stream_stream( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None, _registered_method: Optional[bool] = False, ) -> grpc.StreamStreamMultiCallable: _registered_call_handle = None if _registered_method: _registered_call_handle = self._get_registered_call_handle(method) return _StreamStreamMultiCallable( self._channel, _channel_managed_call_management(self._call_state), _common.encode(method), _common.encode(self._target), request_serializer, response_deserializer, _registered_call_handle, ) def _unsubscribe_all(self) -> None: state = self._connectivity_state if state: with state.lock: del state.callbacks_and_connectivities[:] def _close(self) -> None: self._unsubscribe_all() self._channel.close(cygrpc.StatusCode.cancelled, "Channel closed!") cygrpc.fork_unregister_channel(self) if cygrpc.g_gevent_activated: cygrpc.gevent_decrement_channel_count() def _close_on_fork(self) -> None: self._unsubscribe_all() self._channel.close_on_fork( cygrpc.StatusCode.cancelled, "Channel closed due to fork" ) def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self._close() return False def close(self) -> None: self._close() def __del__(self): # TODO(https://github.com/grpc/grpc/issues/12531): Several releases # after 1.12 (1.16 or thereabouts?) add a "self._channel.close" call # here (or more likely, call self._close() here). We don't do this today # because many valid use cases today allow the channel to be deleted # immediately after stubs are created. After a sufficient period of time # has passed for all users to be trusted to freeze out to their channels # for as long as they are in use and to close them after using them, # then deletion of this grpc._channel.Channel instance can be made to # effect closure of the underlying cygrpc.Channel instance. try: self._unsubscribe_all() except: # pylint: disable=bare-except # Exceptions in __del__ are ignored by Python anyway, but they can # keep spamming logs. Just silence them. pass	79.44 KB	2025-05-25 12:26
📄 _common.py # 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. """Shared implementation.""" import logging import time from typing import Any, AnyStr, Callable, Optional, Union import grpc from grpc._cython import cygrpc from grpc._typing import DeserializingFunction from grpc._typing import SerializingFunction _LOGGER = logging.getLogger(__name__) CYGRPC_CONNECTIVITY_STATE_TO_CHANNEL_CONNECTIVITY = { cygrpc.ConnectivityState.idle: grpc.ChannelConnectivity.IDLE, cygrpc.ConnectivityState.connecting: grpc.ChannelConnectivity.CONNECTING, cygrpc.ConnectivityState.ready: grpc.ChannelConnectivity.READY, cygrpc.ConnectivityState.transient_failure: grpc.ChannelConnectivity.TRANSIENT_FAILURE, cygrpc.ConnectivityState.shutdown: grpc.ChannelConnectivity.SHUTDOWN, } CYGRPC_STATUS_CODE_TO_STATUS_CODE = { cygrpc.StatusCode.ok: grpc.StatusCode.OK, cygrpc.StatusCode.cancelled: grpc.StatusCode.CANCELLED, cygrpc.StatusCode.unknown: grpc.StatusCode.UNKNOWN, cygrpc.StatusCode.invalid_argument: grpc.StatusCode.INVALID_ARGUMENT, cygrpc.StatusCode.deadline_exceeded: grpc.StatusCode.DEADLINE_EXCEEDED, cygrpc.StatusCode.not_found: grpc.StatusCode.NOT_FOUND, cygrpc.StatusCode.already_exists: grpc.StatusCode.ALREADY_EXISTS, cygrpc.StatusCode.permission_denied: grpc.StatusCode.PERMISSION_DENIED, cygrpc.StatusCode.unauthenticated: grpc.StatusCode.UNAUTHENTICATED, cygrpc.StatusCode.resource_exhausted: grpc.StatusCode.RESOURCE_EXHAUSTED, cygrpc.StatusCode.failed_precondition: grpc.StatusCode.FAILED_PRECONDITION, cygrpc.StatusCode.aborted: grpc.StatusCode.ABORTED, cygrpc.StatusCode.out_of_range: grpc.StatusCode.OUT_OF_RANGE, cygrpc.StatusCode.unimplemented: grpc.StatusCode.UNIMPLEMENTED, cygrpc.StatusCode.internal: grpc.StatusCode.INTERNAL, cygrpc.StatusCode.unavailable: grpc.StatusCode.UNAVAILABLE, cygrpc.StatusCode.data_loss: grpc.StatusCode.DATA_LOSS, } STATUS_CODE_TO_CYGRPC_STATUS_CODE = { grpc_code: cygrpc_code for cygrpc_code, grpc_code in CYGRPC_STATUS_CODE_TO_STATUS_CODE.items() } MAXIMUM_WAIT_TIMEOUT = 0.1 _ERROR_MESSAGE_PORT_BINDING_FAILED = ( "Failed to bind to address %s; set " "GRPC_VERBOSITY=debug environment variable to see detailed error message." ) def encode(s: AnyStr) -> bytes: if isinstance(s, bytes): return s else: return s.encode("utf8") def decode(b: AnyStr) -> str: if isinstance(b, bytes): return b.decode("utf-8", "replace") return b def _transform( message: Any, transformer: Union[SerializingFunction, DeserializingFunction, None], exception_message: str, ) -> Any: if transformer is None: return message else: try: return transformer(message) except Exception: # pylint: disable=broad-except _LOGGER.exception(exception_message) return None def serialize(message: Any, serializer: Optional[SerializingFunction]) -> bytes: return _transform(message, serializer, "Exception serializing message!") def deserialize( serialized_message: bytes, deserializer: Optional[DeserializingFunction] ) -> Any: return _transform( serialized_message, deserializer, "Exception deserializing message!" ) def fully_qualified_method(group: str, method: str) -> str: return "/{}/{}".format(group, method) def _wait_once( wait_fn: Callable[..., bool], timeout: float, spin_cb: Optional[Callable[[], None]], ): wait_fn(timeout=timeout) if spin_cb is not None: spin_cb() def wait( wait_fn: Callable[..., bool], wait_complete_fn: Callable[[], bool], timeout: Optional[float] = None, spin_cb: Optional[Callable[[], None]] = None, ) -> bool: """Blocks waiting for an event without blocking the thread indefinitely. See https://github.com/grpc/grpc/issues/19464 for full context. CPython's `threading.Event.wait` and `threading.Condition.wait` methods, if invoked without a timeout kwarg, may block the calling thread indefinitely. If the call is made from the main thread, this means that signal handlers may not run for an arbitrarily long period of time. This wrapper calls the supplied wait function with an arbitrary short timeout to ensure that no signal handler has to wait longer than MAXIMUM_WAIT_TIMEOUT before executing. Args: wait_fn: A callable acceptable a single float-valued kwarg named `timeout`. This function is expected to be one of `threading.Event.wait` or `threading.Condition.wait`. wait_complete_fn: A callable taking no arguments and returning a bool. When this function returns true, it indicates that waiting should cease. timeout: An optional float-valued number of seconds after which the wait should cease. spin_cb: An optional Callable taking no arguments and returning nothing. This callback will be called on each iteration of the spin. This may be used for, e.g. work related to forking. Returns: True if a timeout was supplied and it was reached. False otherwise. """ if timeout is None: while not wait_complete_fn(): _wait_once(wait_fn, MAXIMUM_WAIT_TIMEOUT, spin_cb) else: end = time.time() + timeout while not wait_complete_fn(): remaining = min(end - time.time(), MAXIMUM_WAIT_TIMEOUT) if remaining < 0: return True _wait_once(wait_fn, remaining, spin_cb) return False def validate_port_binding_result(address: str, port: int) -> int: """Validates if the port binding succeed. If the port returned by Core is 0, the binding is failed. However, in that case, the Core API doesn't return a detailed failing reason. The best we can do is raising an exception to prevent further confusion. Args: address: The address string to be bound. port: An int returned by core """ if port == 0: # The Core API doesn't return a failure message. The best we can do # is raising an exception to prevent further confusion. raise RuntimeError(_ERROR_MESSAGE_PORT_BINDING_FAILED % address) else: return port	6.63 KB	2025-05-25 12:26
📄 _compression.py # Copyright 2019 The 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. from __future__ import annotations from typing import Optional import grpc from grpc._cython import cygrpc from grpc._typing import MetadataType NoCompression = cygrpc.CompressionAlgorithm.none Deflate = cygrpc.CompressionAlgorithm.deflate Gzip = cygrpc.CompressionAlgorithm.gzip _METADATA_STRING_MAPPING = { NoCompression: "identity", Deflate: "deflate", Gzip: "gzip", } def _compression_algorithm_to_metadata_value( compression: grpc.Compression, ) -> str: return _METADATA_STRING_MAPPING[compression] def compression_algorithm_to_metadata(compression: grpc.Compression): return ( cygrpc.GRPC_COMPRESSION_REQUEST_ALGORITHM_MD_KEY, _compression_algorithm_to_metadata_value(compression), ) def create_channel_option(compression: Optional[grpc.Compression]): return ( ((cygrpc.GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM, int(compression)),) if compression else () ) def augment_metadata( metadata: Optional[MetadataType], compression: Optional[grpc.Compression] ): if not metadata and not compression: return None base_metadata = tuple(metadata) if metadata else () compression_metadata = ( (compression_algorithm_to_metadata(compression),) if compression else () ) return base_metadata + compression_metadata __all__ = ( "NoCompression", "Deflate", "Gzip", )	1.94 KB	2025-05-25 12:26
📁 _cython	-	2025-05-25 12:26
📄 _grpcio_metadata.py __version__ = """1.71.0"""	26 B	2025-05-25 12:26
📄 _interceptor.py # Copyright 2017 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. """Implementation of gRPC Python interceptors.""" import collections import sys import types from typing import Any, Callable, Optional, Sequence, Tuple, Union import grpc from ._typing import DeserializingFunction from ._typing import DoneCallbackType from ._typing import MetadataType from ._typing import RequestIterableType from ._typing import SerializingFunction class _ServicePipeline(object): interceptors: Tuple[grpc.ServerInterceptor] def __init__(self, interceptors: Sequence[grpc.ServerInterceptor]): self.interceptors = tuple(interceptors) def _continuation(self, thunk: Callable, index: int) -> Callable: return lambda context: self._intercept_at(thunk, index, context) def _intercept_at( self, thunk: Callable, index: int, context: grpc.HandlerCallDetails ) -> grpc.RpcMethodHandler: if index < len(self.interceptors): interceptor = self.interceptors[index] thunk = self._continuation(thunk, index + 1) return interceptor.intercept_service(thunk, context) else: return thunk(context) def execute( self, thunk: Callable, context: grpc.HandlerCallDetails ) -> grpc.RpcMethodHandler: return self._intercept_at(thunk, 0, context) def service_pipeline( interceptors: Optional[Sequence[grpc.ServerInterceptor]], ) -> Optional[_ServicePipeline]: return _ServicePipeline(interceptors) if interceptors else None class _ClientCallDetails( collections.namedtuple( "_ClientCallDetails", ( "method", "timeout", "metadata", "credentials", "wait_for_ready", "compression", ), ), grpc.ClientCallDetails, ): pass def _unwrap_client_call_details( call_details: grpc.ClientCallDetails, default_details: grpc.ClientCallDetails, ) -> Tuple[ str, float, MetadataType, grpc.CallCredentials, bool, grpc.Compression ]: try: method = call_details.method # pytype: disable=attribute-error except AttributeError: method = default_details.method # pytype: disable=attribute-error try: timeout = call_details.timeout # pytype: disable=attribute-error except AttributeError: timeout = default_details.timeout # pytype: disable=attribute-error try: metadata = call_details.metadata # pytype: disable=attribute-error except AttributeError: metadata = default_details.metadata # pytype: disable=attribute-error try: credentials = ( call_details.credentials ) # pytype: disable=attribute-error except AttributeError: credentials = ( default_details.credentials ) # pytype: disable=attribute-error try: wait_for_ready = ( call_details.wait_for_ready ) # pytype: disable=attribute-error except AttributeError: wait_for_ready = ( default_details.wait_for_ready ) # pytype: disable=attribute-error try: compression = ( call_details.compression ) # pytype: disable=attribute-error except AttributeError: compression = ( default_details.compression ) # pytype: disable=attribute-error return method, timeout, metadata, credentials, wait_for_ready, compression class _FailureOutcome( grpc.RpcError, grpc.Future, grpc.Call ): # pylint: disable=too-many-ancestors _exception: Exception _traceback: types.TracebackType def __init__(self, exception: Exception, traceback: types.TracebackType): super(_FailureOutcome, self).__init__() self._exception = exception self._traceback = traceback def initial_metadata(self) -> Optional[MetadataType]: return None def trailing_metadata(self) -> Optional[MetadataType]: return None def code(self) -> Optional[grpc.StatusCode]: return grpc.StatusCode.INTERNAL def details(self) -> Optional[str]: return "Exception raised while intercepting the RPC" def cancel(self) -> bool: return False def cancelled(self) -> bool: return False def is_active(self) -> bool: return False def time_remaining(self) -> Optional[float]: return None def running(self) -> bool: return False def done(self) -> bool: return True def result(self, ignored_timeout: Optional[float] = None): raise self._exception def exception( self, ignored_timeout: Optional[float] = None ) -> Optional[Exception]: return self._exception def traceback( self, ignored_timeout: Optional[float] = None ) -> Optional[types.TracebackType]: return self._traceback def add_callback(self, unused_callback) -> bool: return False def add_done_callback(self, fn: DoneCallbackType) -> None: fn(self) def __iter__(self): return self def __next__(self): raise self._exception def next(self): return self.__next__() class _UnaryOutcome(grpc.Call, grpc.Future): _response: Any _call: grpc.Call def __init__(self, response: Any, call: grpc.Call): self._response = response self._call = call def initial_metadata(self) -> Optional[MetadataType]: return self._call.initial_metadata() def trailing_metadata(self) -> Optional[MetadataType]: return self._call.trailing_metadata() def code(self) -> Optional[grpc.StatusCode]: return self._call.code() def details(self) -> Optional[str]: return self._call.details() def is_active(self) -> bool: return self._call.is_active() def time_remaining(self) -> Optional[float]: return self._call.time_remaining() def cancel(self) -> bool: return self._call.cancel() def add_callback(self, callback) -> bool: return self._call.add_callback(callback) def cancelled(self) -> bool: return False def running(self) -> bool: return False def done(self) -> bool: return True def result(self, ignored_timeout: Optional[float] = None): return self._response def exception(self, ignored_timeout: Optional[float] = None): return None def traceback(self, ignored_timeout: Optional[float] = None): return None def add_done_callback(self, fn: DoneCallbackType) -> None: fn(self) class _UnaryUnaryMultiCallable(grpc.UnaryUnaryMultiCallable): _thunk: Callable _method: str _interceptor: grpc.UnaryUnaryClientInterceptor def __init__( self, thunk: Callable, method: str, interceptor: grpc.UnaryUnaryClientInterceptor, ): self._thunk = thunk self._method = method self._interceptor = interceptor def __call__( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Any: response, ignored_call = self._with_call( request, timeout=timeout, metadata=metadata, credentials=credentials, wait_for_ready=wait_for_ready, compression=compression, ) return response def _with_call( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Tuple[Any, grpc.Call]: client_call_details = _ClientCallDetails( self._method, timeout, metadata, credentials, wait_for_ready, compression, ) def continuation(new_details, request): ( new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready, new_compression, ) = _unwrap_client_call_details(new_details, client_call_details) try: response, call = self._thunk(new_method).with_call( request, timeout=new_timeout, metadata=new_metadata, credentials=new_credentials, wait_for_ready=new_wait_for_ready, compression=new_compression, ) return _UnaryOutcome(response, call) except grpc.RpcError as rpc_error: return rpc_error except Exception as exception: # pylint:disable=broad-except return _FailureOutcome(exception, sys.exc_info()[2]) call = self._interceptor.intercept_unary_unary( continuation, client_call_details, request ) return call.result(), call def with_call( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Tuple[Any, grpc.Call]: return self._with_call( request, timeout=timeout, metadata=metadata, credentials=credentials, wait_for_ready=wait_for_ready, compression=compression, ) def future( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Any: client_call_details = _ClientCallDetails( self._method, timeout, metadata, credentials, wait_for_ready, compression, ) def continuation(new_details, request): ( new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready, new_compression, ) = _unwrap_client_call_details(new_details, client_call_details) return self._thunk(new_method).future( request, timeout=new_timeout, metadata=new_metadata, credentials=new_credentials, wait_for_ready=new_wait_for_ready, compression=new_compression, ) try: return self._interceptor.intercept_unary_unary( continuation, client_call_details, request ) except Exception as exception: # pylint:disable=broad-except return _FailureOutcome(exception, sys.exc_info()[2]) class _UnaryStreamMultiCallable(grpc.UnaryStreamMultiCallable): _thunk: Callable _method: str _interceptor: grpc.UnaryStreamClientInterceptor def __init__( self, thunk: Callable, method: str, interceptor: grpc.UnaryStreamClientInterceptor, ): self._thunk = thunk self._method = method self._interceptor = interceptor def __call__( self, request: Any, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ): client_call_details = _ClientCallDetails( self._method, timeout, metadata, credentials, wait_for_ready, compression, ) def continuation(new_details, request): ( new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready, new_compression, ) = _unwrap_client_call_details(new_details, client_call_details) return self._thunk(new_method)( request, timeout=new_timeout, metadata=new_metadata, credentials=new_credentials, wait_for_ready=new_wait_for_ready, compression=new_compression, ) try: return self._interceptor.intercept_unary_stream( continuation, client_call_details, request ) except Exception as exception: # pylint:disable=broad-except return _FailureOutcome(exception, sys.exc_info()[2]) class _StreamUnaryMultiCallable(grpc.StreamUnaryMultiCallable): _thunk: Callable _method: str _interceptor: grpc.StreamUnaryClientInterceptor def __init__( self, thunk: Callable, method: str, interceptor: grpc.StreamUnaryClientInterceptor, ): self._thunk = thunk self._method = method self._interceptor = interceptor def __call__( self, request_iterator: RequestIterableType, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Any: response, ignored_call = self._with_call( request_iterator, timeout=timeout, metadata=metadata, credentials=credentials, wait_for_ready=wait_for_ready, compression=compression, ) return response def _with_call( self, request_iterator: RequestIterableType, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Tuple[Any, grpc.Call]: client_call_details = _ClientCallDetails( self._method, timeout, metadata, credentials, wait_for_ready, compression, ) def continuation(new_details, request_iterator): ( new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready, new_compression, ) = _unwrap_client_call_details(new_details, client_call_details) try: response, call = self._thunk(new_method).with_call( request_iterator, timeout=new_timeout, metadata=new_metadata, credentials=new_credentials, wait_for_ready=new_wait_for_ready, compression=new_compression, ) return _UnaryOutcome(response, call) except grpc.RpcError as rpc_error: return rpc_error except Exception as exception: # pylint:disable=broad-except return _FailureOutcome(exception, sys.exc_info()[2]) call = self._interceptor.intercept_stream_unary( continuation, client_call_details, request_iterator ) return call.result(), call def with_call( self, request_iterator: RequestIterableType, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Tuple[Any, grpc.Call]: return self._with_call( request_iterator, timeout=timeout, metadata=metadata, credentials=credentials, wait_for_ready=wait_for_ready, compression=compression, ) def future( self, request_iterator: RequestIterableType, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ) -> Any: client_call_details = _ClientCallDetails( self._method, timeout, metadata, credentials, wait_for_ready, compression, ) def continuation(new_details, request_iterator): ( new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready, new_compression, ) = _unwrap_client_call_details(new_details, client_call_details) return self._thunk(new_method).future( request_iterator, timeout=new_timeout, metadata=new_metadata, credentials=new_credentials, wait_for_ready=new_wait_for_ready, compression=new_compression, ) try: return self._interceptor.intercept_stream_unary( continuation, client_call_details, request_iterator ) except Exception as exception: # pylint:disable=broad-except return _FailureOutcome(exception, sys.exc_info()[2]) class _StreamStreamMultiCallable(grpc.StreamStreamMultiCallable): _thunk: Callable _method: str _interceptor: grpc.StreamStreamClientInterceptor def __init__( self, thunk: Callable, method: str, interceptor: grpc.StreamStreamClientInterceptor, ): self._thunk = thunk self._method = method self._interceptor = interceptor def __call__( self, request_iterator: RequestIterableType, timeout: Optional[float] = None, metadata: Optional[MetadataType] = None, credentials: Optional[grpc.CallCredentials] = None, wait_for_ready: Optional[bool] = None, compression: Optional[grpc.Compression] = None, ): client_call_details = _ClientCallDetails( self._method, timeout, metadata, credentials, wait_for_ready, compression, ) def continuation(new_details, request_iterator): ( new_method, new_timeout, new_metadata, new_credentials, new_wait_for_ready, new_compression, ) = _unwrap_client_call_details(new_details, client_call_details) return self._thunk(new_method)( request_iterator, timeout=new_timeout, metadata=new_metadata, credentials=new_credentials, wait_for_ready=new_wait_for_ready, compression=new_compression, ) try: return self._interceptor.intercept_stream_stream( continuation, client_call_details, request_iterator ) except Exception as exception: # pylint:disable=broad-except return _FailureOutcome(exception, sys.exc_info()[2]) class _Channel(grpc.Channel): _channel: grpc.Channel _interceptor: Union[ grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor, grpc.StreamStreamClientInterceptor, grpc.StreamUnaryClientInterceptor, ] def __init__( self, channel: grpc.Channel, interceptor: Union[ grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor, grpc.StreamStreamClientInterceptor, grpc.StreamUnaryClientInterceptor, ], ): self._channel = channel self._interceptor = interceptor def subscribe( self, callback: Callable, try_to_connect: Optional[bool] = False ): self._channel.subscribe(callback, try_to_connect=try_to_connect) def unsubscribe(self, callback: Callable): self._channel.unsubscribe(callback) # pylint: disable=arguments-differ def unary_unary( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None, _registered_method: Optional[bool] = False, ) -> grpc.UnaryUnaryMultiCallable: # pytype: disable=wrong-arg-count thunk = lambda m: self._channel.unary_unary( m, request_serializer, response_deserializer, _registered_method, ) # pytype: enable=wrong-arg-count if isinstance(self._interceptor, grpc.UnaryUnaryClientInterceptor): return _UnaryUnaryMultiCallable(thunk, method, self._interceptor) else: return thunk(method) # pylint: disable=arguments-differ def unary_stream( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None, _registered_method: Optional[bool] = False, ) -> grpc.UnaryStreamMultiCallable: # pytype: disable=wrong-arg-count thunk = lambda m: self._channel.unary_stream( m, request_serializer, response_deserializer, _registered_method, ) # pytype: enable=wrong-arg-count if isinstance(self._interceptor, grpc.UnaryStreamClientInterceptor): return _UnaryStreamMultiCallable(thunk, method, self._interceptor) else: return thunk(method) # pylint: disable=arguments-differ def stream_unary( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None, _registered_method: Optional[bool] = False, ) -> grpc.StreamUnaryMultiCallable: # pytype: disable=wrong-arg-count thunk = lambda m: self._channel.stream_unary( m, request_serializer, response_deserializer, _registered_method, ) # pytype: enable=wrong-arg-count if isinstance(self._interceptor, grpc.StreamUnaryClientInterceptor): return _StreamUnaryMultiCallable(thunk, method, self._interceptor) else: return thunk(method) # pylint: disable=arguments-differ def stream_stream( self, method: str, request_serializer: Optional[SerializingFunction] = None, response_deserializer: Optional[DeserializingFunction] = None, _registered_method: Optional[bool] = False, ) -> grpc.StreamStreamMultiCallable: # pytype: disable=wrong-arg-count thunk = lambda m: self._channel.stream_stream( m, request_serializer, response_deserializer, _registered_method, ) # pytype: enable=wrong-arg-count if isinstance(self._interceptor, grpc.StreamStreamClientInterceptor): return _StreamStreamMultiCallable(thunk, method, self._interceptor) else: return thunk(method) def _close(self): self._channel.close() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self._close() return False def close(self): self._channel.close() def intercept_channel( channel: grpc.Channel, *interceptors: Optional[ Sequence[ Union[ grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor, grpc.StreamStreamClientInterceptor, grpc.StreamUnaryClientInterceptor, ] ] ], ) -> grpc.Channel: for interceptor in reversed(list(interceptors)): if ( not isinstance(interceptor, grpc.UnaryUnaryClientInterceptor) and not isinstance(interceptor, grpc.UnaryStreamClientInterceptor) and not isinstance(interceptor, grpc.StreamUnaryClientInterceptor) and not isinstance(interceptor, grpc.StreamStreamClientInterceptor) ): raise TypeError( "interceptor must be " "grpc.UnaryUnaryClientInterceptor or " "grpc.UnaryStreamClientInterceptor or " "grpc.StreamUnaryClientInterceptor or " "grpc.StreamStreamClientInterceptor or " ) channel = _Channel(channel, interceptor) return channel	25.26 KB	2025-05-25 12:26
📄 _observability.py # Copyright 2023 The 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. from __future__ import annotations import abc import contextlib import logging import threading from typing import Any, Generator, Generic, List, Optional, TypeVar from grpc._cython import cygrpc as _cygrpc from grpc._typing import ChannelArgumentType _LOGGER = logging.getLogger(__name__) _channel = Any # _channel.py imports this module. ClientCallTracerCapsule = TypeVar("ClientCallTracerCapsule") ServerCallTracerFactoryCapsule = TypeVar("ServerCallTracerFactoryCapsule") _plugin_lock: threading.RLock = threading.RLock() _OBSERVABILITY_PLUGIN: Optional["ObservabilityPlugin"] = None _SERVICES_TO_EXCLUDE: List[bytes] = [ b"google.monitoring.v3.MetricService", b"google.devtools.cloudtrace.v2.TraceService", ] class ServerCallTracerFactory: """An encapsulation of a ServerCallTracerFactory. Instances of this class can be passed to a Channel as values for the grpc.experimental.server_call_tracer_factory option """ def __init__(self, address): self._address = address def __int__(self): return self._address class ObservabilityPlugin( Generic[ClientCallTracerCapsule, ServerCallTracerFactoryCapsule], metaclass=abc.ABCMeta, ): """Abstract base class for observability plugin. This is a semi-private class that was intended for the exclusive use of the gRPC team. The ClientCallTracerCapsule and ClientCallTracerCapsule created by this plugin should be injected to gRPC core using observability_init at the start of a program, before any channels/servers are built. Any future methods added to this interface cannot have the @abc.abstractmethod annotation. Attributes: _stats_enabled: A bool indicates whether tracing is enabled. _tracing_enabled: A bool indicates whether stats(metrics) is enabled. _registered_methods: A set which stores the registered method names in bytes. """ _tracing_enabled: bool = False _stats_enabled: bool = False @abc.abstractmethod def create_client_call_tracer( self, method_name: bytes, target: bytes ) -> ClientCallTracerCapsule: """Creates a ClientCallTracerCapsule. After register the plugin, if tracing or stats is enabled, this method will be called after a call was created, the ClientCallTracer created by this method will be saved to call context. The ClientCallTracer is an object which implements `grpc_core::ClientCallTracer` interface and wrapped in a PyCapsule using `client_call_tracer` as name. Args: method_name: The method name of the call in byte format. target: The channel target of the call in byte format. registered_method: Whether this method is pre-registered. Returns: A PyCapsule which stores a ClientCallTracer object. """ raise NotImplementedError() @abc.abstractmethod def save_trace_context( self, trace_id: str, span_id: str, is_sampled: bool ) -> None: """Saves the trace_id and span_id related to the current span. After register the plugin, if tracing is enabled, this method will be called after the server finished sending response. This method can be used to propagate census context. Args: trace_id: The identifier for the trace associated with the span as a 32-character hexadecimal encoded string, e.g. 26ed0036f2eff2b7317bccce3e28d01f span_id: The identifier for the span as a 16-character hexadecimal encoded string. e.g. 113ec879e62583bc is_sampled: A bool indicates whether the span is sampled. """ raise NotImplementedError() @abc.abstractmethod def create_server_call_tracer_factory( self, , xds: bool = False, ) -> Optional[ServerCallTracerFactoryCapsule]: """Creates a ServerCallTracerFactoryCapsule. This method will be called at server initialization time to create a ServerCallTracerFactory, which will be registered to gRPC core. The ServerCallTracerFactory is an object which implements `grpc_core::ServerCallTracerFactory` interface and wrapped in a PyCapsule using `server_call_tracer_factory` as name. Args: xds: Whether the server is xds server. Returns: A PyCapsule which stores a ServerCallTracerFactory object. Or None if plugin decides not to create ServerCallTracerFactory. """ raise NotImplementedError() @abc.abstractmethod def record_rpc_latency( self, method: str, target: str, rpc_latency: float, status_code: Any ) -> None: """Record the latency of the RPC. After register the plugin, if stats is enabled, this method will be called at the end of each RPC. Args: method: The fully-qualified name of the RPC method being invoked. target: The target name of the RPC method being invoked. rpc_latency: The latency for the RPC in seconds, equals to the time between when the client invokes the RPC and when the client receives the status. status_code: An element of grpc.StatusCode in string format representing the final status for the RPC. """ raise NotImplementedError() def set_tracing(self, enable: bool) -> None: """Enable or disable tracing. Args: enable: A bool indicates whether tracing should be enabled. """ self._tracing_enabled = enable def set_stats(self, enable: bool) -> None: """Enable or disable stats(metrics). Args: enable: A bool indicates whether stats should be enabled. """ self._stats_enabled = enable def save_registered_method(self, method_name: bytes) -> None: """Saves the method name to registered_method list. When exporting metrics, method name for unregistered methods will be replaced with 'other' by default. Args: method_name: The method name in bytes. """ raise NotImplementedError() @property def tracing_enabled(self) -> bool: return self._tracing_enabled @property def stats_enabled(self) -> bool: return self._stats_enabled @property def observability_enabled(self) -> bool: return self.tracing_enabled or self.stats_enabled @contextlib.contextmanager def get_plugin() -> Generator[Optional[ObservabilityPlugin], None, None]: """Get the ObservabilityPlugin in _observability module. Returns: The ObservabilityPlugin currently registered with the _observability module. Or None if no plugin exists at the time of calling this method. """ with _plugin_lock: yield _OBSERVABILITY_PLUGIN def set_plugin(observability_plugin: Optional[ObservabilityPlugin]) -> None: """Save ObservabilityPlugin to _observability module. Args: observability_plugin: The ObservabilityPlugin to save. Raises: ValueError: If an ObservabilityPlugin was already registered at the time of calling this method. """ global _OBSERVABILITY_PLUGIN # pylint: disable=global-statement with _plugin_lock: if observability_plugin and _OBSERVABILITY_PLUGIN: raise ValueError("observability_plugin was already set!") _OBSERVABILITY_PLUGIN = observability_plugin def observability_init(observability_plugin: ObservabilityPlugin) -> None: """Initialize observability with provided ObservabilityPlugin. This method have to be called at the start of a program, before any channels/servers are built. Args: observability_plugin: The ObservabilityPlugin to use. Raises: ValueError: If an ObservabilityPlugin was already registered at the time of calling this method. """ set_plugin(observability_plugin) def observability_deinit() -> None: """Clear the observability context, including ObservabilityPlugin and ServerCallTracerFactory This method have to be called after exit observability context so that it's possible to re-initialize again. """ set_plugin(None) _cygrpc.clear_server_call_tracer_factory() def maybe_record_rpc_latency(state: "_channel._RPCState") -> None: """Record the latency of the RPC, if the plugin is registered and stats is enabled. This method will be called at the end of each RPC. Args: state: a grpc._channel._RPCState object which contains the stats related to the RPC. """ # TODO(xuanwn): use channel args to exclude those metrics. for exclude_prefix in _SERVICES_TO_EXCLUDE: if exclude_prefix in state.method.encode("utf8"): return with get_plugin() as plugin: if plugin and plugin.stats_enabled: rpc_latency_s = state.rpc_end_time - state.rpc_start_time rpc_latency_ms = rpc_latency_s 1000 plugin.record_rpc_latency( state.method, state.target, rpc_latency_ms, state.code ) def create_server_call_tracer_factory_option(xds: bool) -> ChannelArgumentType: with get_plugin() as plugin: if plugin and plugin.stats_enabled: server_call_tracer_factory_address = ( _cygrpc.get_server_call_tracer_factory_address(plugin, xds) ) if server_call_tracer_factory_address: return ( ( "grpc.experimental.server_call_tracer_factory", ServerCallTracerFactory( server_call_tracer_factory_address ), ), ) return ()	10.17 KB	2025-05-25 12:26
📄 _plugin_wrapping.py # Copyright 2015 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 collections import logging import threading from typing import Callable, Optional, Type import grpc from grpc import _common from grpc._cython import cygrpc from grpc._typing import MetadataType _LOGGER = logging.getLogger(__name__) class _AuthMetadataContext( collections.namedtuple( "AuthMetadataContext", ( "service_url", "method_name", ), ), grpc.AuthMetadataContext, ): pass class _CallbackState(object): def __init__(self): self.lock = threading.Lock() self.called = False self.exception = None class _AuthMetadataPluginCallback(grpc.AuthMetadataPluginCallback): _state: _CallbackState _callback: Callable def __init__(self, state: _CallbackState, callback: Callable): self._state = state self._callback = callback def __call__( self, metadata: MetadataType, error: Optional[Type[BaseException]] ): with self._state.lock: if self._state.exception is None: if self._state.called: raise RuntimeError( "AuthMetadataPluginCallback invoked more than once!" ) else: self._state.called = True else: raise RuntimeError( 'AuthMetadataPluginCallback raised exception "{}"!'.format( self._state.exception ) ) if error is None: self._callback(metadata, cygrpc.StatusCode.ok, None) else: self._callback( None, cygrpc.StatusCode.internal, _common.encode(str(error)) ) class _Plugin(object): _metadata_plugin: grpc.AuthMetadataPlugin def __init__(self, metadata_plugin: grpc.AuthMetadataPlugin): self._metadata_plugin = metadata_plugin self._stored_ctx = None try: import contextvars # pylint: disable=wrong-import-position # The plugin may be invoked on a thread created by Core, which will not # have the context propagated. This context is stored and installed in # the thread invoking the plugin. self._stored_ctx = contextvars.copy_context() except ImportError: # Support versions predating contextvars. pass def __call__(self, service_url: str, method_name: str, callback: Callable): context = _AuthMetadataContext( _common.decode(service_url), _common.decode(method_name) ) callback_state = _CallbackState() try: self._metadata_plugin( context, _AuthMetadataPluginCallback(callback_state, callback) ) except Exception as exception: # pylint: disable=broad-except _LOGGER.exception( 'AuthMetadataPluginCallback "%s" raised exception!', self._metadata_plugin, ) with callback_state.lock: callback_state.exception = exception if callback_state.called: return callback( None, cygrpc.StatusCode.internal, _common.encode(str(exception)) ) def metadata_plugin_call_credentials( metadata_plugin: grpc.AuthMetadataPlugin, name: Optional[str] ) -> grpc.CallCredentials: if name is None: try: effective_name = metadata_plugin.__name__ except AttributeError: effective_name = metadata_plugin.__class__.__name__ else: effective_name = name return grpc.CallCredentials( cygrpc.MetadataPluginCallCredentials( _Plugin(metadata_plugin), _common.encode(effective_name) ) )	4.28 KB	2025-05-25 12:26
📄 _runtime_protos.py # Copyright 2020 The 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 sys import types from typing import Tuple, Union _REQUIRED_SYMBOLS = ("_protos", "_services", "_protos_and_services") _MINIMUM_VERSION = (3, 5, 0) _UNINSTALLED_TEMPLATE = ( "Install the grpcio-tools package (1.32.0+) to use the {} function." ) _VERSION_ERROR_TEMPLATE = ( "The {} function is only on available on Python 3.X interpreters." ) def _has_runtime_proto_symbols(mod: types.ModuleType) -> bool: return all(hasattr(mod, sym) for sym in _REQUIRED_SYMBOLS) def _is_grpc_tools_importable() -> bool: try: import grpc_tools # pylint: disable=unused-import # pytype: disable=import-error return True except ImportError as e: # NOTE: It's possible that we're encountering a transitive ImportError, so # we check for that and re-raise if so. if "grpc_tools" not in e.args[0]: raise return False def _call_with_lazy_import( fn_name: str, protobuf_path: str ) -> Union[types.ModuleType, Tuple[types.ModuleType, types.ModuleType]]: """Calls one of the three functions, lazily importing grpc_tools. Args: fn_name: The name of the function to import from grpc_tools.protoc. protobuf_path: The path to import. Returns: The appropriate module object. """ if sys.version_info < _MINIMUM_VERSION: raise NotImplementedError(_VERSION_ERROR_TEMPLATE.format(fn_name)) else: if not _is_grpc_tools_importable(): raise NotImplementedError(_UNINSTALLED_TEMPLATE.format(fn_name)) import grpc_tools.protoc # pytype: disable=import-error if _has_runtime_proto_symbols(grpc_tools.protoc): fn = getattr(grpc_tools.protoc, "_" + fn_name) return fn(protobuf_path) else: raise NotImplementedError(_UNINSTALLED_TEMPLATE.format(fn_name)) def protos(protobuf_path): # pylint: disable=unused-argument """Returns a module generated by the indicated .proto file. THIS IS AN EXPERIMENTAL API. Use this function to retrieve classes corresponding to message definitions in the .proto file. To inspect the contents of the returned module, use the dir function. For example: ``` protos = grpc.protos("foo.proto") print(dir(protos)) ``` The returned module object corresponds to the _pb2.py file generated by protoc. The path is expected to be relative to an entry on sys.path and all transitive dependencies of the file should also be resolvable from an entry on sys.path. To completely disable the machinery behind this function, set the GRPC_PYTHON_DISABLE_DYNAMIC_STUBS environment variable to "true". Args: protobuf_path: The path to the .proto file on the filesystem. This path must be resolvable from an entry on sys.path and so must all of its transitive dependencies. Returns: A module object corresponding to the message code for the indicated .proto file. Equivalent to a generated _pb2.py file. """ return _call_with_lazy_import("protos", protobuf_path) def services(protobuf_path): # pylint: disable=unused-argument """Returns a module generated by the indicated .proto file. THIS IS AN EXPERIMENTAL API. Use this function to retrieve classes and functions corresponding to service definitions in the .proto file, including both stub and servicer definitions. To inspect the contents of the returned module, use the dir function. For example: ``` services = grpc.services("foo.proto") print(dir(services)) ``` The returned module object corresponds to the _pb2_grpc.py file generated by protoc. The path is expected to be relative to an entry on sys.path and all transitive dependencies of the file should also be resolvable from an entry on sys.path. To completely disable the machinery behind this function, set the GRPC_PYTHON_DISABLE_DYNAMIC_STUBS environment variable to "true". Args: protobuf_path: The path to the .proto file on the filesystem. This path must be resolvable from an entry on sys.path and so must all of its transitive dependencies. Returns: A module object corresponding to the stub/service code for the indicated .proto file. Equivalent to a generated _pb2_grpc.py file. """ return _call_with_lazy_import("services", protobuf_path) def protos_and_services(protobuf_path): # pylint: disable=unused-argument """Returns a 2-tuple of modules corresponding to protos and services. THIS IS AN EXPERIMENTAL API. The return value of this function is equivalent to a call to protos and a call to services. To completely disable the machinery behind this function, set the GRPC_PYTHON_DISABLE_DYNAMIC_STUBS environment variable to "true". Args: protobuf_path: The path to the .proto file on the filesystem. This path must be resolvable from an entry on sys.path and so must all of its transitive dependencies. Returns: A 2-tuple of module objects corresponding to (protos(path), services(path)). """ return _call_with_lazy_import("protos_and_services", protobuf_path)	5.67 KB	2025-05-25 12:26
📄 _server.py # 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. """Service-side implementation of gRPC Python.""" from __future__ import annotations import abc import collections from concurrent import futures import contextvars import enum import logging import threading import time import traceback from typing import ( Any, Callable, Dict, Iterable, Iterator, List, Mapping, Optional, Sequence, Set, Tuple, Union, ) import grpc # pytype: disable=pyi-error from grpc import _common # pytype: disable=pyi-error from grpc import _compression # pytype: disable=pyi-error from grpc import _interceptor # pytype: disable=pyi-error from grpc import _observability # pytype: disable=pyi-error from grpc._cython import cygrpc from grpc._typing import ArityAgnosticMethodHandler from grpc._typing import ChannelArgumentType from grpc._typing import DeserializingFunction from grpc._typing import MetadataType from grpc._typing import NullaryCallbackType from grpc._typing import ResponseType from grpc._typing import SerializingFunction from grpc._typing import ServerCallbackTag from grpc._typing import ServerTagCallbackType _LOGGER = logging.getLogger(__name__) _SHUTDOWN_TAG = "shutdown" _REQUEST_CALL_TAG = "request_call" _RECEIVE_CLOSE_ON_SERVER_TOKEN = "receive_close_on_server" _SEND_INITIAL_METADATA_TOKEN = "send_initial_metadata" _RECEIVE_MESSAGE_TOKEN = "receive_message" _SEND_MESSAGE_TOKEN = "send_message" _SEND_INITIAL_METADATA_AND_SEND_MESSAGE_TOKEN = ( "send_initial_metadata * send_message" ) _SEND_STATUS_FROM_SERVER_TOKEN = "send_status_from_server" _SEND_INITIAL_METADATA_AND_SEND_STATUS_FROM_SERVER_TOKEN = ( "send_initial_metadata * send_status_from_server" ) _OPEN = "open" _CLOSED = "closed" _CANCELLED = "cancelled" _EMPTY_FLAGS = 0 _DEALLOCATED_SERVER_CHECK_PERIOD_S = 1.0 _INF_TIMEOUT = 1e9 def _serialized_request(request_event: cygrpc.BaseEvent) -> bytes: return request_event.batch_operations[0].message() def _application_code(code: grpc.StatusCode) -> cygrpc.StatusCode: cygrpc_code = _common.STATUS_CODE_TO_CYGRPC_STATUS_CODE.get(code) return cygrpc.StatusCode.unknown if cygrpc_code is None else cygrpc_code def _completion_code(state: _RPCState) -> cygrpc.StatusCode: if state.code is None: return cygrpc.StatusCode.ok else: return _application_code(state.code) def _abortion_code( state: _RPCState, code: cygrpc.StatusCode ) -> cygrpc.StatusCode: if state.code is None: return code else: return _application_code(state.code) def _details(state: _RPCState) -> bytes: return b"" if state.details is None else state.details class _HandlerCallDetails( collections.namedtuple( "_HandlerCallDetails", ( "method", "invocation_metadata", ), ), grpc.HandlerCallDetails, ): pass class _Method(abc.ABC): @abc.abstractmethod def name(self) -> Optional[str]: raise NotImplementedError() @abc.abstractmethod def handler( self, handler_call_details: _HandlerCallDetails ) -> Optional[grpc.RpcMethodHandler]: raise NotImplementedError() class _RegisteredMethod(_Method): def __init__( self, name: str, registered_handler: Optional[grpc.RpcMethodHandler], ): self._name = name self._registered_handler = registered_handler def name(self) -> Optional[str]: return self._name def handler( self, handler_call_details: _HandlerCallDetails ) -> Optional[grpc.RpcMethodHandler]: return self._registered_handler class _GenericMethod(_Method): def __init__( self, generic_handlers: List[grpc.GenericRpcHandler], ): self._generic_handlers = generic_handlers def name(self) -> Optional[str]: return None def handler( self, handler_call_details: _HandlerCallDetails ) -> Optional[grpc.RpcMethodHandler]: # If the same method have both generic and registered handler, # registered handler will take precedence. for generic_handler in self._generic_handlers: method_handler = generic_handler.service(handler_call_details) if method_handler is not None: return method_handler return None class _RPCState(object): context: contextvars.Context condition: threading.Condition due = Set[str] request: Any client: str initial_metadata_allowed: bool compression_algorithm: Optional[grpc.Compression] disable_next_compression: bool trailing_metadata: Optional[MetadataType] code: Optional[grpc.StatusCode] details: Optional[bytes] statused: bool rpc_errors: List[Exception] callbacks: Optional[List[NullaryCallbackType]] aborted: bool def __init__(self): self.context = contextvars.Context() self.condition = threading.Condition() self.due = set() self.request = None self.client = _OPEN self.initial_metadata_allowed = True self.compression_algorithm = None self.disable_next_compression = False self.trailing_metadata = None self.code = None self.details = None self.statused = False self.rpc_errors = [] self.callbacks = [] self.aborted = False def _raise_rpc_error(state: _RPCState) -> None: rpc_error = grpc.RpcError() state.rpc_errors.append(rpc_error) raise rpc_error def _possibly_finish_call( state: _RPCState, token: str ) -> ServerTagCallbackType: state.due.remove(token) if not _is_rpc_state_active(state) and not state.due: callbacks = state.callbacks state.callbacks = None return state, callbacks else: return None, () def _send_status_from_server(state: _RPCState, token: str) -> ServerCallbackTag: def send_status_from_server(unused_send_status_from_server_event): with state.condition: return _possibly_finish_call(state, token) return send_status_from_server def _get_initial_metadata( state: _RPCState, metadata: Optional[MetadataType] ) -> Optional[MetadataType]: with state.condition: if state.compression_algorithm: compression_metadata = ( _compression.compression_algorithm_to_metadata( state.compression_algorithm ), ) if metadata is None: return compression_metadata else: return compression_metadata + tuple(metadata) else: return metadata def _get_initial_metadata_operation( state: _RPCState, metadata: Optional[MetadataType] ) -> cygrpc.Operation: operation = cygrpc.SendInitialMetadataOperation( _get_initial_metadata(state, metadata), _EMPTY_FLAGS ) return operation def _abort( state: _RPCState, call: cygrpc.Call, code: cygrpc.StatusCode, details: bytes ) -> None: if state.client is not _CANCELLED: effective_code = _abortion_code(state, code) effective_details = details if state.details is None else state.details if state.initial_metadata_allowed: operations = ( _get_initial_metadata_operation(state, None), cygrpc.SendStatusFromServerOperation( state.trailing_metadata, effective_code, effective_details, _EMPTY_FLAGS, ), ) token = _SEND_INITIAL_METADATA_AND_SEND_STATUS_FROM_SERVER_TOKEN else: operations = ( cygrpc.SendStatusFromServerOperation( state.trailing_metadata, effective_code, effective_details, _EMPTY_FLAGS, ), ) token = _SEND_STATUS_FROM_SERVER_TOKEN call.start_server_batch( operations, _send_status_from_server(state, token) ) state.statused = True state.due.add(token) def _receive_close_on_server(state: _RPCState) -> ServerCallbackTag: def receive_close_on_server(receive_close_on_server_event): with state.condition: if receive_close_on_server_event.batch_operations[0].cancelled(): state.client = _CANCELLED elif state.client is _OPEN: state.client = _CLOSED state.condition.notify_all() return _possibly_finish_call(state, _RECEIVE_CLOSE_ON_SERVER_TOKEN) return receive_close_on_server def _receive_message( state: _RPCState, call: cygrpc.Call, request_deserializer: Optional[DeserializingFunction], ) -> ServerCallbackTag: def receive_message(receive_message_event): serialized_request = _serialized_request(receive_message_event) if serialized_request is None: with state.condition: if state.client is _OPEN: state.client = _CLOSED state.condition.notify_all() return _possibly_finish_call(state, _RECEIVE_MESSAGE_TOKEN) else: request = _common.deserialize( serialized_request, request_deserializer ) with state.condition: if request is None: _abort( state, call, cygrpc.StatusCode.internal, b"Exception deserializing request!", ) else: state.request = request state.condition.notify_all() return _possibly_finish_call(state, _RECEIVE_MESSAGE_TOKEN) return receive_message def _send_initial_metadata(state: _RPCState) -> ServerCallbackTag: def send_initial_metadata(unused_send_initial_metadata_event): with state.condition: return _possibly_finish_call(state, _SEND_INITIAL_METADATA_TOKEN) return send_initial_metadata def _send_message(state: _RPCState, token: str) -> ServerCallbackTag: def send_message(unused_send_message_event): with state.condition: state.condition.notify_all() return _possibly_finish_call(state, token) return send_message class _Context(grpc.ServicerContext): _rpc_event: cygrpc.BaseEvent _state: _RPCState request_deserializer: Optional[DeserializingFunction] def __init__( self, rpc_event: cygrpc.BaseEvent, state: _RPCState, request_deserializer: Optional[DeserializingFunction], ): self._rpc_event = rpc_event self._state = state self._request_deserializer = request_deserializer def is_active(self) -> bool: with self._state.condition: return _is_rpc_state_active(self._state) def time_remaining(self) -> float: return max(self._rpc_event.call_details.deadline - time.time(), 0) def cancel(self) -> None: self._rpc_event.call.cancel() def add_callback(self, callback: NullaryCallbackType) -> bool: with self._state.condition: if self._state.callbacks is None: return False else: self._state.callbacks.append(callback) return True def disable_next_message_compression(self) -> None: with self._state.condition: self._state.disable_next_compression = True def invocation_metadata(self) -> Optional[MetadataType]: return self._rpc_event.invocation_metadata def peer(self) -> str: return _common.decode(self._rpc_event.call.peer()) def peer_identities(self) -> Optional[Sequence[bytes]]: return cygrpc.peer_identities(self._rpc_event.call) def peer_identity_key(self) -> Optional[str]: id_key = cygrpc.peer_identity_key(self._rpc_event.call) return id_key if id_key is None else _common.decode(id_key) def auth_context(self) -> Mapping[str, Sequence[bytes]]: auth_context = cygrpc.auth_context(self._rpc_event.call) auth_context_dict = {} if auth_context is None else auth_context return { _common.decode(key): value for key, value in auth_context_dict.items() } def set_compression(self, compression: grpc.Compression) -> None: with self._state.condition: self._state.compression_algorithm = compression def send_initial_metadata(self, initial_metadata: MetadataType) -> None: with self._state.condition: if self._state.client is _CANCELLED: _raise_rpc_error(self._state) else: if self._state.initial_metadata_allowed: operation = _get_initial_metadata_operation( self._state, initial_metadata ) self._rpc_event.call.start_server_batch( (operation,), _send_initial_metadata(self._state) ) self._state.initial_metadata_allowed = False self._state.due.add(_SEND_INITIAL_METADATA_TOKEN) else: raise ValueError("Initial metadata no longer allowed!") def set_trailing_metadata(self, trailing_metadata: MetadataType) -> None: with self._state.condition: self._state.trailing_metadata = trailing_metadata def trailing_metadata(self) -> Optional[MetadataType]: return self._state.trailing_metadata def abort(self, code: grpc.StatusCode, details: str) -> None: # treat OK like other invalid arguments: fail the RPC if code == grpc.StatusCode.OK: _LOGGER.error( "abort() called with StatusCode.OK; returning UNKNOWN" ) code = grpc.StatusCode.UNKNOWN details = "" with self._state.condition: self._state.code = code self._state.details = _common.encode(details) self._state.aborted = True raise Exception() def abort_with_status(self, status: grpc.Status) -> None: self._state.trailing_metadata = status.trailing_metadata self.abort(status.code, status.details) def set_code(self, code: grpc.StatusCode) -> None: with self._state.condition: self._state.code = code def code(self) -> grpc.StatusCode: return self._state.code def set_details(self, details: str) -> None: with self._state.condition: self._state.details = _common.encode(details) def details(self) -> bytes: return self._state.details def _finalize_state(self) -> None: pass class _RequestIterator(object): _state: _RPCState _call: cygrpc.Call _request_deserializer: Optional[DeserializingFunction] def __init__( self, state: _RPCState, call: cygrpc.Call, request_deserializer: Optional[DeserializingFunction], ): self._state = state self._call = call self._request_deserializer = request_deserializer def _raise_or_start_receive_message(self) -> None: if self._state.client is _CANCELLED: _raise_rpc_error(self._state) elif not _is_rpc_state_active(self._state): raise StopIteration() else: self._call.start_server_batch( (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),), _receive_message( self._state, self._call, self._request_deserializer ), ) self._state.due.add(_RECEIVE_MESSAGE_TOKEN) def _look_for_request(self) -> Any: if self._state.client is _CANCELLED: _raise_rpc_error(self._state) elif ( self._state.request is None and _RECEIVE_MESSAGE_TOKEN not in self._state.due ): raise StopIteration() else: request = self._state.request self._state.request = None return request raise AssertionError() # should never run def _next(self) -> Any: with self._state.condition: self._raise_or_start_receive_message() while True: self._state.condition.wait() request = self._look_for_request() if request is not None: return request def __iter__(self) -> _RequestIterator: return self def __next__(self) -> Any: return self._next() def next(self) -> Any: return self._next() def _unary_request( rpc_event: cygrpc.BaseEvent, state: _RPCState, request_deserializer: Optional[DeserializingFunction], ) -> Callable[[], Any]: def unary_request(): with state.condition: if not _is_rpc_state_active(state): return None else: rpc_event.call.start_server_batch( (cygrpc.ReceiveMessageOperation(_EMPTY_FLAGS),), _receive_message( state, rpc_event.call, request_deserializer ), ) state.due.add(_RECEIVE_MESSAGE_TOKEN) while True: state.condition.wait() if state.request is None: if state.client is _CLOSED: details = '"{}" requires exactly one request message.'.format( rpc_event.call_details.method ) _abort( state, rpc_event.call, cygrpc.StatusCode.unimplemented, _common.encode(details), ) return None elif state.client is _CANCELLED: return None else: request = state.request state.request = None return request return unary_request def _call_behavior( rpc_event: cygrpc.BaseEvent, state: _RPCState, behavior: ArityAgnosticMethodHandler, argument: Any, request_deserializer: Optional[DeserializingFunction], send_response_callback: Optional[Callable[[ResponseType], None]] = None, ) -> Tuple[Union[ResponseType, Iterator[ResponseType]], bool]: from grpc import _create_servicer_context # pytype: disable=pyi-error with _create_servicer_context( rpc_event, state, request_deserializer ) as context: try: response_or_iterator = None if send_response_callback is not None: response_or_iterator = behavior( argument, context, send_response_callback ) else: response_or_iterator = behavior(argument, context) return response_or_iterator, True except Exception as exception: # pylint: disable=broad-except with state.condition: if state.aborted: _abort( state, rpc_event.call, cygrpc.StatusCode.unknown, b"RPC Aborted", ) elif exception not in state.rpc_errors: try: details = "Exception calling application: {}".format( exception ) except Exception: # pylint: disable=broad-except details = ( "Calling application raised unprintable Exception!" ) _LOGGER.exception( traceback.format_exception( type(exception), exception, exception.__traceback__, ) ) traceback.print_exc() _LOGGER.exception(details) _abort( state, rpc_event.call, cygrpc.StatusCode.unknown, _common.encode(details), ) return None, False def _take_response_from_response_iterator( rpc_event: cygrpc.BaseEvent, state: _RPCState, response_iterator: Iterator[ResponseType], ) -> Tuple[ResponseType, bool]: try: return next(response_iterator), True except StopIteration: return None, True except Exception as exception: # pylint: disable=broad-except with state.condition: if state.aborted: _abort( state, rpc_event.call, cygrpc.StatusCode.unknown, b"RPC Aborted", ) elif exception not in state.rpc_errors: details = "Exception iterating responses: {}".format(exception) _LOGGER.exception(details) _abort( state, rpc_event.call, cygrpc.StatusCode.unknown, _common.encode(details), ) return None, False def _serialize_response( rpc_event: cygrpc.BaseEvent, state: _RPCState, response: Any, response_serializer: Optional[SerializingFunction], ) -> Optional[bytes]: serialized_response = _common.serialize(response, response_serializer) if serialized_response is None: with state.condition: _abort( state, rpc_event.call, cygrpc.StatusCode.internal, b"Failed to serialize response!", ) return None else: return serialized_response def _get_send_message_op_flags_from_state( state: _RPCState, ) -> Union[int, cygrpc.WriteFlag]: if state.disable_next_compression: return cygrpc.WriteFlag.no_compress else: return _EMPTY_FLAGS def _reset_per_message_state(state: _RPCState) -> None: with state.condition: state.disable_next_compression = False def _send_response( rpc_event: cygrpc.BaseEvent, state: _RPCState, serialized_response: bytes ) -> bool: with state.condition: if not _is_rpc_state_active(state): return False else: if state.initial_metadata_allowed: operations = ( _get_initial_metadata_operation(state, None), cygrpc.SendMessageOperation( serialized_response, _get_send_message_op_flags_from_state(state), ), ) state.initial_metadata_allowed = False token = _SEND_INITIAL_METADATA_AND_SEND_MESSAGE_TOKEN else: operations = ( cygrpc.SendMessageOperation( serialized_response, _get_send_message_op_flags_from_state(state), ), ) token = _SEND_MESSAGE_TOKEN rpc_event.call.start_server_batch( operations, _send_message(state, token) ) state.due.add(token) _reset_per_message_state(state) while True: state.condition.wait() if token not in state.due: return _is_rpc_state_active(state) def _status( rpc_event: cygrpc.BaseEvent, state: _RPCState, serialized_response: Optional[bytes], ) -> None: with state.condition: if state.client is not _CANCELLED: code = _completion_code(state) details = _details(state) operations = [ cygrpc.SendStatusFromServerOperation( state.trailing_metadata, code, details, _EMPTY_FLAGS ), ] if state.initial_metadata_allowed: operations.append(_get_initial_metadata_operation(state, None)) if serialized_response is not None: operations.append( cygrpc.SendMessageOperation( serialized_response, _get_send_message_op_flags_from_state(state), ) ) rpc_event.call.start_server_batch( operations, _send_status_from_server(state, _SEND_STATUS_FROM_SERVER_TOKEN), ) state.statused = True _reset_per_message_state(state) state.due.add(_SEND_STATUS_FROM_SERVER_TOKEN) def _unary_response_in_pool( rpc_event: cygrpc.BaseEvent, state: _RPCState, behavior: ArityAgnosticMethodHandler, argument_thunk: Callable[[], Any], request_deserializer: Optional[SerializingFunction], response_serializer: Optional[SerializingFunction], ) -> None: cygrpc.install_context_from_request_call_event(rpc_event) try: argument = argument_thunk() if argument is not None: response, proceed = _call_behavior( rpc_event, state, behavior, argument, request_deserializer ) if proceed: serialized_response = _serialize_response( rpc_event, state, response, response_serializer ) if serialized_response is not None: _status(rpc_event, state, serialized_response) except Exception: # pylint: disable=broad-except traceback.print_exc() finally: cygrpc.uninstall_context() def _stream_response_in_pool( rpc_event: cygrpc.BaseEvent, state: _RPCState, behavior: ArityAgnosticMethodHandler, argument_thunk: Callable[[], Any], request_deserializer: Optional[DeserializingFunction], response_serializer: Optional[SerializingFunction], ) -> None: cygrpc.install_context_from_request_call_event(rpc_event) def send_response(response: Any) -> None: if response is None: _status(rpc_event, state, None) else: serialized_response = _serialize_response( rpc_event, state, response, response_serializer ) if serialized_response is not None: _send_response(rpc_event, state, serialized_response) try: argument = argument_thunk() if argument is not None: if ( hasattr(behavior, "experimental_non_blocking") and behavior.experimental_non_blocking ): _call_behavior( rpc_event, state, behavior, argument, request_deserializer, send_response_callback=send_response, ) else: response_iterator, proceed = _call_behavior( rpc_event, state, behavior, argument, request_deserializer ) if proceed: _send_message_callback_to_blocking_iterator_adapter( rpc_event, state, send_response, response_iterator ) except Exception: # pylint: disable=broad-except traceback.print_exc() finally: cygrpc.uninstall_context() def _is_rpc_state_active(state: _RPCState) -> bool: return state.client is not _CANCELLED and not state.statused def _send_message_callback_to_blocking_iterator_adapter( rpc_event: cygrpc.BaseEvent, state: _RPCState, send_response_callback: Callable[[ResponseType], None], response_iterator: Iterator[ResponseType], ) -> None: while True: response, proceed = _take_response_from_response_iterator( rpc_event, state, response_iterator ) if proceed: send_response_callback(response) if not _is_rpc_state_active(state): break else: break def _select_thread_pool_for_behavior( behavior: ArityAgnosticMethodHandler, default_thread_pool: futures.ThreadPoolExecutor, ) -> futures.ThreadPoolExecutor: if hasattr(behavior, "experimental_thread_pool") and isinstance( behavior.experimental_thread_pool, futures.ThreadPoolExecutor ): return behavior.experimental_thread_pool else: return default_thread_pool def _handle_unary_unary( rpc_event: cygrpc.BaseEvent, state: _RPCState, method_handler: grpc.RpcMethodHandler, default_thread_pool: futures.ThreadPoolExecutor, ) -> futures.Future: unary_request = _unary_request( rpc_event, state, method_handler.request_deserializer ) thread_pool = _select_thread_pool_for_behavior( method_handler.unary_unary, default_thread_pool ) return thread_pool.submit( state.context.run, _unary_response_in_pool, rpc_event, state, method_handler.unary_unary, unary_request, method_handler.request_deserializer, method_handler.response_serializer, ) def _handle_unary_stream( rpc_event: cygrpc.BaseEvent, state: _RPCState, method_handler: grpc.RpcMethodHandler, default_thread_pool: futures.ThreadPoolExecutor, ) -> futures.Future: unary_request = _unary_request( rpc_event, state, method_handler.request_deserializer ) thread_pool = _select_thread_pool_for_behavior( method_handler.unary_stream, default_thread_pool ) return thread_pool.submit( state.context.run, _stream_response_in_pool, rpc_event, state, method_handler.unary_stream, unary_request, method_handler.request_deserializer, method_handler.response_serializer, ) def _handle_stream_unary( rpc_event: cygrpc.BaseEvent, state: _RPCState, method_handler: grpc.RpcMethodHandler, default_thread_pool: futures.ThreadPoolExecutor, ) -> futures.Future: request_iterator = _RequestIterator( state, rpc_event.call, method_handler.request_deserializer ) thread_pool = _select_thread_pool_for_behavior( method_handler.stream_unary, default_thread_pool ) return thread_pool.submit( state.context.run, _unary_response_in_pool, rpc_event, state, method_handler.stream_unary, lambda: request_iterator, method_handler.request_deserializer, method_handler.response_serializer, ) def _handle_stream_stream( rpc_event: cygrpc.BaseEvent, state: _RPCState, method_handler: grpc.RpcMethodHandler, default_thread_pool: futures.ThreadPoolExecutor, ) -> futures.Future: request_iterator = _RequestIterator( state, rpc_event.call, method_handler.request_deserializer ) thread_pool = _select_thread_pool_for_behavior( method_handler.stream_stream, default_thread_pool ) return thread_pool.submit( state.context.run, _stream_response_in_pool, rpc_event, state, method_handler.stream_stream, lambda: request_iterator, method_handler.request_deserializer, method_handler.response_serializer, ) def _find_method_handler( rpc_event: cygrpc.BaseEvent, state: _RPCState, method_with_handler: _Method, interceptor_pipeline: Optional[_interceptor._ServicePipeline], ) -> Optional[grpc.RpcMethodHandler]: def query_handlers( handler_call_details: _HandlerCallDetails, ) -> Optional[grpc.RpcMethodHandler]: return method_with_handler.handler(handler_call_details) method_name = method_with_handler.name() if not method_name: method_name = _common.decode(rpc_event.call_details.method) handler_call_details = _HandlerCallDetails( method_name, rpc_event.invocation_metadata, ) if interceptor_pipeline is not None: return state.context.run( interceptor_pipeline.execute, query_handlers, handler_call_details ) else: return state.context.run(query_handlers, handler_call_details) def _reject_rpc( rpc_event: cygrpc.BaseEvent, rpc_state: _RPCState, status: cygrpc.StatusCode, details: bytes, ): operations = ( _get_initial_metadata_operation(rpc_state, None), cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS), cygrpc.SendStatusFromServerOperation( None, status, details, _EMPTY_FLAGS ), ) rpc_event.call.start_server_batch( operations, lambda ignored_event: ( rpc_state, (), ), ) def _handle_with_method_handler( rpc_event: cygrpc.BaseEvent, state: _RPCState, method_handler: grpc.RpcMethodHandler, thread_pool: futures.ThreadPoolExecutor, ) -> futures.Future: with state.condition: rpc_event.call.start_server_batch( (cygrpc.ReceiveCloseOnServerOperation(_EMPTY_FLAGS),), _receive_close_on_server(state), ) state.due.add(_RECEIVE_CLOSE_ON_SERVER_TOKEN) if method_handler.request_streaming: if method_handler.response_streaming: return _handle_stream_stream( rpc_event, state, method_handler, thread_pool ) else: return _handle_stream_unary( rpc_event, state, method_handler, thread_pool ) else: if method_handler.response_streaming: return _handle_unary_stream( rpc_event, state, method_handler, thread_pool ) else: return _handle_unary_unary( rpc_event, state, method_handler, thread_pool ) def _handle_call( rpc_event: cygrpc.BaseEvent, method_with_handler: _Method, interceptor_pipeline: Optional[_interceptor._ServicePipeline], thread_pool: futures.ThreadPoolExecutor, concurrency_exceeded: bool, ) -> Tuple[Optional[_RPCState], Optional[futures.Future]]: """Handles RPC based on provided handlers. When receiving a call event from Core, registered method will have its name as tag, we pass the tag as registered_method_name to this method, then we can find the handler in registered_method_handlers based on the method name. For call event with unregistered method, the method name will be included in rpc_event.call_details.method and we need to query the generics handlers to find the actual handler. """ if not rpc_event.success: return None, None if rpc_event.call_details.method or method_with_handler.name(): rpc_state = _RPCState() try: method_handler = _find_method_handler( rpc_event, rpc_state, method_with_handler, interceptor_pipeline, ) except Exception as exception: # pylint: disable=broad-except details = "Exception servicing handler: {}".format(exception) _LOGGER.exception(details) _reject_rpc( rpc_event, rpc_state, cygrpc.StatusCode.unknown, b"Error in service handler!", ) return rpc_state, None if method_handler is None: _reject_rpc( rpc_event, rpc_state, cygrpc.StatusCode.unimplemented, b"Method not found!", ) return rpc_state, None elif concurrency_exceeded: _reject_rpc( rpc_event, rpc_state, cygrpc.StatusCode.resource_exhausted, b"Concurrent RPC limit exceeded!", ) return rpc_state, None else: return ( rpc_state, _handle_with_method_handler( rpc_event, rpc_state, method_handler, thread_pool ), ) else: return None, None @enum.unique class _ServerStage(enum.Enum): STOPPED = "stopped" STARTED = "started" GRACE = "grace" class _ServerState(object): lock: threading.RLock completion_queue: cygrpc.CompletionQueue server: cygrpc.Server generic_handlers: List[grpc.GenericRpcHandler] registered_method_handlers: Dict[str, grpc.RpcMethodHandler] interceptor_pipeline: Optional[_interceptor._ServicePipeline] thread_pool: futures.ThreadPoolExecutor stage: _ServerStage termination_event: threading.Event shutdown_events: List[threading.Event] maximum_concurrent_rpcs: Optional[int] active_rpc_count: int rpc_states: Set[_RPCState] due: Set[str] server_deallocated: bool # pylint: disable=too-many-arguments def __init__( self, completion_queue: cygrpc.CompletionQueue, server: cygrpc.Server, generic_handlers: Sequence[grpc.GenericRpcHandler], interceptor_pipeline: Optional[_interceptor._ServicePipeline], thread_pool: futures.ThreadPoolExecutor, maximum_concurrent_rpcs: Optional[int], ): self.lock = threading.RLock() self.completion_queue = completion_queue self.server = server self.generic_handlers = list(generic_handlers) self.interceptor_pipeline = interceptor_pipeline self.thread_pool = thread_pool self.stage = _ServerStage.STOPPED self.termination_event = threading.Event() self.shutdown_events = [self.termination_event] self.maximum_concurrent_rpcs = maximum_concurrent_rpcs self.active_rpc_count = 0 self.registered_method_handlers = {} # TODO(https://github.com/grpc/grpc/issues/6597): eliminate these fields. self.rpc_states = set() self.due = set() # A "volatile" flag to interrupt the daemon serving thread self.server_deallocated = False def _add_generic_handlers( state: _ServerState, generic_handlers: Iterable[grpc.GenericRpcHandler] ) -> None: with state.lock: state.generic_handlers.extend(generic_handlers) def _add_registered_method_handlers( state: _ServerState, method_handlers: Dict[str, grpc.RpcMethodHandler] ) -> None: with state.lock: state.registered_method_handlers.update(method_handlers) def _add_insecure_port(state: _ServerState, address: bytes) -> int: with state.lock: return state.server.add_http2_port(address) def _add_secure_port( state: _ServerState, address: bytes, server_credentials: grpc.ServerCredentials, ) -> int: with state.lock: return state.server.add_http2_port( address, server_credentials._credentials ) def _request_call(state: _ServerState) -> None: state.server.request_call( state.completion_queue, state.completion_queue, _REQUEST_CALL_TAG ) state.due.add(_REQUEST_CALL_TAG) def _request_registered_call(state: _ServerState, method: str) -> None: registered_call_tag = method state.server.request_registered_call( state.completion_queue, state.completion_queue, method, registered_call_tag, ) state.due.add(registered_call_tag) # TODO(https://github.com/grpc/grpc/issues/6597): delete this function. def _stop_serving(state: _ServerState) -> bool: if not state.rpc_states and not state.due: state.server.destroy() for shutdown_event in state.shutdown_events: shutdown_event.set() state.stage = _ServerStage.STOPPED return True else: return False def _on_call_completed(state: _ServerState) -> None: with state.lock: state.active_rpc_count -= 1 # pylint: disable=too-many-branches def _process_event_and_continue( state: _ServerState, event: cygrpc.BaseEvent ) -> bool: should_continue = True if event.tag is _SHUTDOWN_TAG: with state.lock: state.due.remove(_SHUTDOWN_TAG) if _stop_serving(state): should_continue = False elif ( event.tag is _REQUEST_CALL_TAG or event.tag in state.registered_method_handlers.keys() ): registered_method_name = None if event.tag in state.registered_method_handlers.keys(): registered_method_name = event.tag method_with_handler = _RegisteredMethod( registered_method_name, state.registered_method_handlers.get( registered_method_name, None ), ) else: method_with_handler = _GenericMethod( state.generic_handlers, ) with state.lock: state.due.remove(event.tag) concurrency_exceeded = ( state.maximum_concurrent_rpcs is not None and state.active_rpc_count >= state.maximum_concurrent_rpcs ) rpc_state, rpc_future = _handle_call( event, method_with_handler, state.interceptor_pipeline, state.thread_pool, concurrency_exceeded, ) if rpc_state is not None: state.rpc_states.add(rpc_state) if rpc_future is not None: state.active_rpc_count += 1 rpc_future.add_done_callback( lambda unused_future: _on_call_completed(state) ) if state.stage is _ServerStage.STARTED: if ( registered_method_name in state.registered_method_handlers.keys() ): _request_registered_call(state, registered_method_name) else: _request_call(state) elif _stop_serving(state): should_continue = False else: rpc_state, callbacks = event.tag(event) for callback in callbacks: try: callback() except Exception: # pylint: disable=broad-except _LOGGER.exception("Exception calling callback!") if rpc_state is not None: with state.lock: state.rpc_states.remove(rpc_state) if _stop_serving(state): should_continue = False return should_continue def _serve(state: _ServerState) -> None: while True: timeout = time.time() + _DEALLOCATED_SERVER_CHECK_PERIOD_S event = state.completion_queue.poll(timeout) if state.server_deallocated: _begin_shutdown_once(state) if event.completion_type != cygrpc.CompletionType.queue_timeout: if not _process_event_and_continue(state, event): return # We want to force the deletion of the previous event # ~before~ we poll again; if the event has a reference # to a shutdown Call object, this can induce spinlock. event = None def _begin_shutdown_once(state: _ServerState) -> None: with state.lock: if state.stage is _ServerStage.STARTED: state.server.shutdown(state.completion_queue, _SHUTDOWN_TAG) state.stage = _ServerStage.GRACE state.due.add(_SHUTDOWN_TAG) def _stop(state: _ServerState, grace: Optional[float]) -> threading.Event: with state.lock: if state.stage is _ServerStage.STOPPED: shutdown_event = threading.Event() shutdown_event.set() return shutdown_event else: _begin_shutdown_once(state) shutdown_event = threading.Event() state.shutdown_events.append(shutdown_event) if grace is None: state.server.cancel_all_calls() else: def cancel_all_calls_after_grace(): shutdown_event.wait(timeout=grace) with state.lock: state.server.cancel_all_calls() thread = threading.Thread(target=cancel_all_calls_after_grace) thread.start() return shutdown_event shutdown_event.wait() return shutdown_event def _start(state: _ServerState) -> None: with state.lock: if state.stage is not _ServerStage.STOPPED: raise ValueError("Cannot start already-started server!") state.server.start() state.stage = _ServerStage.STARTED # Request a call for each registered method so we can handle any of them. for method in state.registered_method_handlers.keys(): _request_registered_call(state, method) # Also request a call for non-registered method. _request_call(state) thread = threading.Thread(target=_serve, args=(state,)) thread.daemon = True thread.start() def _validate_generic_rpc_handlers( generic_rpc_handlers: Iterable[grpc.GenericRpcHandler], ) -> None: for generic_rpc_handler in generic_rpc_handlers: service_attribute = getattr(generic_rpc_handler, "service", None) if service_attribute is None: raise AttributeError( '"{}" must conform to grpc.GenericRpcHandler type but does ' 'not have "service" method!'.format(generic_rpc_handler) ) def _augment_options( base_options: Sequence[ChannelArgumentType], compression: Optional[grpc.Compression], xds: bool, ) -> Sequence[ChannelArgumentType]: compression_option = _compression.create_channel_option(compression) maybe_server_call_tracer_factory_option = ( _observability.create_server_call_tracer_factory_option(xds) ) return ( tuple(base_options) + compression_option + maybe_server_call_tracer_factory_option ) class _Server(grpc.Server): _state: _ServerState # pylint: disable=too-many-arguments def __init__( self, thread_pool: futures.ThreadPoolExecutor, generic_handlers: Sequence[grpc.GenericRpcHandler], interceptors: Sequence[grpc.ServerInterceptor], options: Sequence[ChannelArgumentType], maximum_concurrent_rpcs: Optional[int], compression: Optional[grpc.Compression], xds: bool, ): completion_queue = cygrpc.CompletionQueue() server = cygrpc.Server(_augment_options(options, compression, xds), xds) server.register_completion_queue(completion_queue) self._state = _ServerState( completion_queue, server, generic_handlers, _interceptor.service_pipeline(interceptors), thread_pool, maximum_concurrent_rpcs, ) self._cy_server = server def add_generic_rpc_handlers( self, generic_rpc_handlers: Iterable[grpc.GenericRpcHandler] ) -> None: _validate_generic_rpc_handlers(generic_rpc_handlers) _add_generic_handlers(self._state, generic_rpc_handlers) def add_registered_method_handlers( self, service_name: str, method_handlers: Dict[str, grpc.RpcMethodHandler], ) -> None: # Can't register method once server started. with self._state.lock: if self._state.stage is _ServerStage.STARTED: return # TODO(xuanwn): We should validate method_handlers first. method_to_handlers = { _common.fully_qualified_method(service_name, method): method_handler for method, method_handler in method_handlers.items() } for fully_qualified_method in method_to_handlers.keys(): self._cy_server.register_method(fully_qualified_method) _add_registered_method_handlers(self._state, method_to_handlers) def add_insecure_port(self, address: str) -> int: return _common.validate_port_binding_result( address, _add_insecure_port(self._state, _common.encode(address)) ) def add_secure_port( self, address: str, server_credentials: grpc.ServerCredentials ) -> int: return _common.validate_port_binding_result( address, _add_secure_port( self._state, _common.encode(address), server_credentials ), ) def start(self) -> None: _start(self._state) def wait_for_termination(self, timeout: Optional[float] = None) -> bool: # NOTE(https://bugs.python.org/issue35935) # Remove this workaround once threading.Event.wait() is working with # CTRL+C across platforms. return _common.wait( self._state.termination_event.wait, self._state.termination_event.is_set, timeout=timeout, ) def stop(self, grace: Optional[float]) -> threading.Event: return _stop(self._state, grace) def __del__(self): if hasattr(self, "_state"): # We can not grab a lock in __del__(), so set a flag to signal the # serving daemon thread (if it exists) to initiate shutdown. self._state.server_deallocated = True def create_server( thread_pool: futures.ThreadPoolExecutor, generic_rpc_handlers: Sequence[grpc.GenericRpcHandler], interceptors: Sequence[grpc.ServerInterceptor], options: Sequence[ChannelArgumentType], maximum_concurrent_rpcs: Optional[int], compression: Optional[grpc.Compression], xds: bool, ) -> _Server: _validate_generic_rpc_handlers(generic_rpc_handlers) return _Server( thread_pool, generic_rpc_handlers, interceptors, options, maximum_concurrent_rpcs, compression, xds, )	49.69 KB	2025-05-25 12:26
📄 _simple_stubs.py # Copyright 2020 The 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. """Functions that obviate explicit stubs and explicit channels.""" import collections import datetime import logging import os import threading from typing import ( Any, AnyStr, Callable, Dict, Iterator, Optional, Sequence, Tuple, TypeVar, Union, ) import grpc from grpc.experimental import experimental_api RequestType = TypeVar("RequestType") ResponseType = TypeVar("ResponseType") OptionsType = Sequence[Tuple[str, str]] CacheKey = Tuple[ str, OptionsType, Optional[grpc.ChannelCredentials], Optional[grpc.Compression], ] _LOGGER = logging.getLogger(__name__) _EVICTION_PERIOD_KEY = "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" if _EVICTION_PERIOD_KEY in os.environ: _EVICTION_PERIOD = datetime.timedelta( seconds=float(os.environ[_EVICTION_PERIOD_KEY]) ) _LOGGER.debug( "Setting managed channel eviction period to %s", _EVICTION_PERIOD ) else: _EVICTION_PERIOD = datetime.timedelta(minutes=10) _MAXIMUM_CHANNELS_KEY = "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" if _MAXIMUM_CHANNELS_KEY in os.environ: _MAXIMUM_CHANNELS = int(os.environ[_MAXIMUM_CHANNELS_KEY]) _LOGGER.debug("Setting maximum managed channels to %d", _MAXIMUM_CHANNELS) else: _MAXIMUM_CHANNELS = 2*8 _DEFAULT_TIMEOUT_KEY = "GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS" if _DEFAULT_TIMEOUT_KEY in os.environ: _DEFAULT_TIMEOUT = float(os.environ[_DEFAULT_TIMEOUT_KEY]) _LOGGER.debug("Setting default timeout seconds to %f", _DEFAULT_TIMEOUT) else: _DEFAULT_TIMEOUT = 60.0 def _create_channel( target: str, options: Sequence[Tuple[str, str]], channel_credentials: Optional[grpc.ChannelCredentials], compression: Optional[grpc.Compression], ) -> grpc.Channel: _LOGGER.debug( f"Creating secure channel with credentials '{channel_credentials}', " + f"options '{options}' and compression '{compression}'" ) return grpc.secure_channel( target, credentials=channel_credentials, options=options, compression=compression, ) class ChannelCache: # NOTE(rbellevi): Untyped due to reference cycle. _singleton = None _lock: threading.RLock = threading.RLock() _condition: threading.Condition = threading.Condition(lock=_lock) _eviction_ready: threading.Event = threading.Event() _mapping: Dict[CacheKey, Tuple[grpc.Channel, datetime.datetime]] _eviction_thread: threading.Thread def __init__(self): self._mapping = collections.OrderedDict() self._eviction_thread = threading.Thread( target=ChannelCache._perform_evictions, daemon=True ) self._eviction_thread.start() @staticmethod def get(): with ChannelCache._lock: if ChannelCache._singleton is None: ChannelCache._singleton = ChannelCache() ChannelCache._eviction_ready.wait() return ChannelCache._singleton def _evict_locked(self, key: CacheKey): channel, _ = self._mapping.pop(key) _LOGGER.debug( "Evicting channel %s with configuration %s.", channel, key ) channel.close() del channel @staticmethod def _perform_evictions(): while True: with ChannelCache._lock: ChannelCache._eviction_ready.set() if not ChannelCache._singleton._mapping: ChannelCache._condition.wait() elif len(ChannelCache._singleton._mapping) > _MAXIMUM_CHANNELS: key = next(iter(ChannelCache._singleton._mapping.keys())) ChannelCache._singleton._evict_locked(key) # And immediately reevaluate. else: key, (_, eviction_time) = next( iter(ChannelCache._singleton._mapping.items()) ) now = datetime.datetime.now() if eviction_time <= now: ChannelCache._singleton._evict_locked(key) continue else: time_to_eviction = (eviction_time - now).total_seconds() # NOTE: We aim to eventually* coalesce to a state in # which no overdue channels are in the cache and the # length of the cache is longer than _MAXIMUM_CHANNELS. # We tolerate momentary states in which these two # criteria are not met. ChannelCache._condition.wait(timeout=time_to_eviction) def get_channel( self, target: str, options: Sequence[Tuple[str, str]], channel_credentials: Optional[grpc.ChannelCredentials], insecure: bool, compression: Optional[grpc.Compression], method: str, _registered_method: bool, ) -> Tuple[grpc.Channel, Optional[int]]: """Get a channel from cache or creates a new channel. This method also takes care of register method for channel, which means we'll register a new call handle if we're calling a non-registered method for an existing channel. Returns: A tuple with two items. The first item is the channel, second item is the call handle if the method is registered, None if it's not registered. """ if insecure and channel_credentials: raise ValueError( "The insecure option is mutually exclusive with " + "the channel_credentials option. Please use one " + "or the other." ) if insecure: channel_credentials = ( grpc.experimental.insecure_channel_credentials() ) elif channel_credentials is None: _LOGGER.debug("Defaulting to SSL channel credentials.") channel_credentials = grpc.ssl_channel_credentials() key = (target, options, channel_credentials, compression) with self._lock: channel_data = self._mapping.get(key, None) call_handle = None if channel_data is not None: channel = channel_data[0] # Register a new call handle if we're calling a registered method for an # existing channel and this method is not registered. if _registered_method: call_handle = channel._get_registered_call_handle(method) self._mapping.pop(key) self._mapping[key] = ( channel, datetime.datetime.now() + _EVICTION_PERIOD, ) return channel, call_handle else: channel = _create_channel( target, options, channel_credentials, compression ) if _registered_method: call_handle = channel._get_registered_call_handle(method) self._mapping[key] = ( channel, datetime.datetime.now() + _EVICTION_PERIOD, ) if ( len(self._mapping) == 1 or len(self._mapping) >= _MAXIMUM_CHANNELS ): self._condition.notify() return channel, call_handle def _test_only_channel_count(self) -> int: with self._lock: return len(self._mapping) @experimental_api # pylint: disable=too-many-locals def unary_unary( request: RequestType, target: str, method: str, request_serializer: Optional[Callable[[Any], bytes]] = None, response_deserializer: Optional[Callable[[bytes], Any]] = None, options: Sequence[Tuple[AnyStr, AnyStr]] = (), channel_credentials: Optional[grpc.ChannelCredentials] = None, insecure: bool = False, call_credentials: Optional[grpc.CallCredentials] = None, compression: Optional[grpc.Compression] = None, wait_for_ready: Optional[bool] = None, timeout: Optional[float] = _DEFAULT_TIMEOUT, metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None, _registered_method: Optional[bool] = False, ) -> ResponseType: """Invokes a unary-unary RPC without an explicitly specified channel. THIS IS AN EXPERIMENTAL API. This is backed by a per-process cache of channels. Channels are evicted from the cache after a fixed period by a background. Channels will also be evicted if more than a configured maximum accumulate. The default eviction period is 10 minutes. One may set the environment variable "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" to configure this. The default maximum number of channels is 256. One may set the environment variable "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" to configure this. Args: request: An iterator that yields request values for the RPC. target: The server address. method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC Core runtime) to configure the channel. channel_credentials: A credential applied to the whole channel, e.g. the return value of grpc.ssl_channel_credentials() or grpc.insecure_channel_credentials(). insecure: If True, specifies channel_credentials as :term:`grpc.insecure_channel_credentials()`. This option is mutually exclusive with the `channel_credentials` option. call_credentials: A call credential applied to each call individually, e.g. the output of grpc.metadata_call_credentials() or grpc.access_token_call_credentials(). compression: An optional value indicating the compression method to be used over the lifetime of the channel, e.g. grpc.Compression.Gzip. wait_for_ready: An optional flag indicating whether the RPC should fail immediately if the connection is not ready at the time the RPC is invoked, or if it should wait until the connection to the server becomes ready. When using this option, the user will likely also want to set a timeout. Defaults to True. timeout: An optional duration of time in seconds to allow for the RPC, after which an exception will be raised. If timeout is unspecified, defaults to a timeout controlled by the GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS environment variable. If that is unset, defaults to 60 seconds. Supply a value of None to indicate that no timeout should be enforced. metadata: Optional metadata to send to the server. Returns: The response to the RPC. """ channel, method_handle = ChannelCache.get().get_channel( target, options, channel_credentials, insecure, compression, method, _registered_method, ) multicallable = channel.unary_unary( method, request_serializer, response_deserializer, method_handle ) wait_for_ready = wait_for_ready if wait_for_ready is not None else True return multicallable( request, metadata=metadata, wait_for_ready=wait_for_ready, credentials=call_credentials, timeout=timeout, ) @experimental_api # pylint: disable=too-many-locals def unary_stream( request: RequestType, target: str, method: str, request_serializer: Optional[Callable[[Any], bytes]] = None, response_deserializer: Optional[Callable[[bytes], Any]] = None, options: Sequence[Tuple[AnyStr, AnyStr]] = (), channel_credentials: Optional[grpc.ChannelCredentials] = None, insecure: bool = False, call_credentials: Optional[grpc.CallCredentials] = None, compression: Optional[grpc.Compression] = None, wait_for_ready: Optional[bool] = None, timeout: Optional[float] = _DEFAULT_TIMEOUT, metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None, _registered_method: Optional[bool] = False, ) -> Iterator[ResponseType]: """Invokes a unary-stream RPC without an explicitly specified channel. THIS IS AN EXPERIMENTAL API. This is backed by a per-process cache of channels. Channels are evicted from the cache after a fixed period by a background. Channels will also be evicted if more than a configured maximum accumulate. The default eviction period is 10 minutes. One may set the environment variable "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" to configure this. The default maximum number of channels is 256. One may set the environment variable "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" to configure this. Args: request: An iterator that yields request values for the RPC. target: The server address. method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC Core runtime) to configure the channel. channel_credentials: A credential applied to the whole channel, e.g. the return value of grpc.ssl_channel_credentials(). insecure: If True, specifies channel_credentials as :term:`grpc.insecure_channel_credentials()`. This option is mutually exclusive with the `channel_credentials` option. call_credentials: A call credential applied to each call individually, e.g. the output of grpc.metadata_call_credentials() or grpc.access_token_call_credentials(). compression: An optional value indicating the compression method to be used over the lifetime of the channel, e.g. grpc.Compression.Gzip. wait_for_ready: An optional flag indicating whether the RPC should fail immediately if the connection is not ready at the time the RPC is invoked, or if it should wait until the connection to the server becomes ready. When using this option, the user will likely also want to set a timeout. Defaults to True. timeout: An optional duration of time in seconds to allow for the RPC, after which an exception will be raised. If timeout is unspecified, defaults to a timeout controlled by the GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS environment variable. If that is unset, defaults to 60 seconds. Supply a value of None to indicate that no timeout should be enforced. metadata: Optional metadata to send to the server. Returns: An iterator of responses. """ channel, method_handle = ChannelCache.get().get_channel( target, options, channel_credentials, insecure, compression, method, _registered_method, ) multicallable = channel.unary_stream( method, request_serializer, response_deserializer, method_handle ) wait_for_ready = wait_for_ready if wait_for_ready is not None else True return multicallable( request, metadata=metadata, wait_for_ready=wait_for_ready, credentials=call_credentials, timeout=timeout, ) @experimental_api # pylint: disable=too-many-locals def stream_unary( request_iterator: Iterator[RequestType], target: str, method: str, request_serializer: Optional[Callable[[Any], bytes]] = None, response_deserializer: Optional[Callable[[bytes], Any]] = None, options: Sequence[Tuple[AnyStr, AnyStr]] = (), channel_credentials: Optional[grpc.ChannelCredentials] = None, insecure: bool = False, call_credentials: Optional[grpc.CallCredentials] = None, compression: Optional[grpc.Compression] = None, wait_for_ready: Optional[bool] = None, timeout: Optional[float] = _DEFAULT_TIMEOUT, metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None, _registered_method: Optional[bool] = False, ) -> ResponseType: """Invokes a stream-unary RPC without an explicitly specified channel. THIS IS AN EXPERIMENTAL API. This is backed by a per-process cache of channels. Channels are evicted from the cache after a fixed period by a background. Channels will also be evicted if more than a configured maximum accumulate. The default eviction period is 10 minutes. One may set the environment variable "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" to configure this. The default maximum number of channels is 256. One may set the environment variable "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" to configure this. Args: request_iterator: An iterator that yields request values for the RPC. target: The server address. method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC Core runtime) to configure the channel. channel_credentials: A credential applied to the whole channel, e.g. the return value of grpc.ssl_channel_credentials(). call_credentials: A call credential applied to each call individually, e.g. the output of grpc.metadata_call_credentials() or grpc.access_token_call_credentials(). insecure: If True, specifies channel_credentials as :term:`grpc.insecure_channel_credentials()`. This option is mutually exclusive with the `channel_credentials` option. compression: An optional value indicating the compression method to be used over the lifetime of the channel, e.g. grpc.Compression.Gzip. wait_for_ready: An optional flag indicating whether the RPC should fail immediately if the connection is not ready at the time the RPC is invoked, or if it should wait until the connection to the server becomes ready. When using this option, the user will likely also want to set a timeout. Defaults to True. timeout: An optional duration of time in seconds to allow for the RPC, after which an exception will be raised. If timeout is unspecified, defaults to a timeout controlled by the GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS environment variable. If that is unset, defaults to 60 seconds. Supply a value of None to indicate that no timeout should be enforced. metadata: Optional metadata to send to the server. Returns: The response to the RPC. """ channel, method_handle = ChannelCache.get().get_channel( target, options, channel_credentials, insecure, compression, method, _registered_method, ) multicallable = channel.stream_unary( method, request_serializer, response_deserializer, method_handle ) wait_for_ready = wait_for_ready if wait_for_ready is not None else True return multicallable( request_iterator, metadata=metadata, wait_for_ready=wait_for_ready, credentials=call_credentials, timeout=timeout, ) @experimental_api # pylint: disable=too-many-locals def stream_stream( request_iterator: Iterator[RequestType], target: str, method: str, request_serializer: Optional[Callable[[Any], bytes]] = None, response_deserializer: Optional[Callable[[bytes], Any]] = None, options: Sequence[Tuple[AnyStr, AnyStr]] = (), channel_credentials: Optional[grpc.ChannelCredentials] = None, insecure: bool = False, call_credentials: Optional[grpc.CallCredentials] = None, compression: Optional[grpc.Compression] = None, wait_for_ready: Optional[bool] = None, timeout: Optional[float] = _DEFAULT_TIMEOUT, metadata: Optional[Sequence[Tuple[str, Union[str, bytes]]]] = None, _registered_method: Optional[bool] = False, ) -> Iterator[ResponseType]: """Invokes a stream-stream RPC without an explicitly specified channel. THIS IS AN EXPERIMENTAL API. This is backed by a per-process cache of channels. Channels are evicted from the cache after a fixed period by a background. Channels will also be evicted if more than a configured maximum accumulate. The default eviction period is 10 minutes. One may set the environment variable "GRPC_PYTHON_MANAGED_CHANNEL_EVICTION_SECONDS" to configure this. The default maximum number of channels is 256. One may set the environment variable "GRPC_PYTHON_MANAGED_CHANNEL_MAXIMUM" to configure this. Args: request_iterator: An iterator that yields request values for the RPC. target: The server address. method: The name of the RPC method. request_serializer: Optional :term:`serializer` for serializing the request message. Request goes unserialized in case None is passed. response_deserializer: Optional :term:`deserializer` for deserializing the response message. Response goes undeserialized in case None is passed. options: An optional list of key-value pairs (:term:`channel_arguments` in gRPC Core runtime) to configure the channel. channel_credentials: A credential applied to the whole channel, e.g. the return value of grpc.ssl_channel_credentials(). call_credentials: A call credential applied to each call individually, e.g. the output of grpc.metadata_call_credentials() or grpc.access_token_call_credentials(). insecure: If True, specifies channel_credentials as :term:`grpc.insecure_channel_credentials()`. This option is mutually exclusive with the `channel_credentials` option. compression: An optional value indicating the compression method to be used over the lifetime of the channel, e.g. grpc.Compression.Gzip. wait_for_ready: An optional flag indicating whether the RPC should fail immediately if the connection is not ready at the time the RPC is invoked, or if it should wait until the connection to the server becomes ready. When using this option, the user will likely also want to set a timeout. Defaults to True. timeout: An optional duration of time in seconds to allow for the RPC, after which an exception will be raised. If timeout is unspecified, defaults to a timeout controlled by the GRPC_PYTHON_DEFAULT_TIMEOUT_SECONDS environment variable. If that is unset, defaults to 60 seconds. Supply a value of None to indicate that no timeout should be enforced. metadata: Optional metadata to send to the server. Returns: An iterator of responses. """ channel, method_handle = ChannelCache.get().get_channel( target, options, channel_credentials, insecure, compression, method, _registered_method, ) multicallable = channel.stream_stream( method, request_serializer, response_deserializer, method_handle ) wait_for_ready = wait_for_ready if wait_for_ready is not None else True return multicallable( request_iterator, metadata=metadata, wait_for_ready=wait_for_ready, credentials=call_credentials, timeout=timeout, )	24.03 KB	2025-05-25 12:26
📄 _typing.py # Copyright 2022 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. """Common types for gRPC Sync API""" from typing import ( TYPE_CHECKING, Any, Callable, Iterable, Iterator, Optional, Sequence, Tuple, TypeVar, Union, ) from grpc._cython import cygrpc if TYPE_CHECKING: from grpc import ServicerContext from grpc._server import _RPCState RequestType = TypeVar("RequestType") ResponseType = TypeVar("ResponseType") SerializingFunction = Callable[[Any], bytes] DeserializingFunction = Callable[[bytes], Any] MetadataType = Sequence[Tuple[str, Union[str, bytes]]] ChannelArgumentType = Tuple[str, Any] DoneCallbackType = Callable[[Any], None] NullaryCallbackType = Callable[[], None] RequestIterableType = Iterable[Any] ResponseIterableType = Iterable[Any] UserTag = Callable[[cygrpc.BaseEvent], bool] IntegratedCallFactory = Callable[ [ int, bytes, None, Optional[float], Optional[MetadataType], Optional[cygrpc.CallCredentials], Sequence[Sequence[cygrpc.Operation]], UserTag, Any, ], cygrpc.IntegratedCall, ] ServerTagCallbackType = Tuple[ Optional["_RPCState"], Sequence[NullaryCallbackType] ] ServerCallbackTag = Callable[[cygrpc.BaseEvent], ServerTagCallbackType] ArityAgnosticMethodHandler = Union[ Callable[ [RequestType, "ServicerContext", Callable[[ResponseType], None]], ResponseType, ], Callable[ [RequestType, "ServicerContext", Callable[[ResponseType], None]], Iterator[ResponseType], ], Callable[ [ Iterator[RequestType], "ServicerContext", Callable[[ResponseType], None], ], ResponseType, ], Callable[ [ Iterator[RequestType], "ServicerContext", Callable[[ResponseType], None], ], Iterator[ResponseType], ], Callable[[RequestType, "ServicerContext"], ResponseType], Callable[[RequestType, "ServicerContext"], Iterator[ResponseType]], Callable[[Iterator[RequestType], "ServicerContext"], ResponseType], Callable[ [Iterator[RequestType], "ServicerContext"], Iterator[ResponseType] ], ]	2.69 KB	2025-05-25 12:26
📄 _utilities.py # Copyright 2015 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. """Internal utilities for gRPC Python.""" import collections import logging import threading import time from typing import Callable, Dict, Optional, Sequence import grpc # pytype: disable=pyi-error from grpc import _common # pytype: disable=pyi-error from grpc._typing import DoneCallbackType _LOGGER = logging.getLogger(__name__) _DONE_CALLBACK_EXCEPTION_LOG_MESSAGE = ( 'Exception calling connectivity future "done" callback!' ) class RpcMethodHandler( collections.namedtuple( "_RpcMethodHandler", ( "request_streaming", "response_streaming", "request_deserializer", "response_serializer", "unary_unary", "unary_stream", "stream_unary", "stream_stream", ), ), grpc.RpcMethodHandler, ): pass class DictionaryGenericHandler(grpc.ServiceRpcHandler): _name: str _method_handlers: Dict[str, grpc.RpcMethodHandler] def __init__( self, service: str, method_handlers: Dict[str, grpc.RpcMethodHandler] ): self._name = service self._method_handlers = { _common.fully_qualified_method(service, method): method_handler for method, method_handler in method_handlers.items() } def service_name(self) -> str: return self._name def service( self, handler_call_details: grpc.HandlerCallDetails ) -> Optional[grpc.RpcMethodHandler]: details_method = handler_call_details.method return self._method_handlers.get( details_method ) # pytype: disable=attribute-error class _ChannelReadyFuture(grpc.Future): _condition: threading.Condition _channel: grpc.Channel _matured: bool _cancelled: bool _done_callbacks: Sequence[Callable] def __init__(self, channel: grpc.Channel): self._condition = threading.Condition() self._channel = channel self._matured = False self._cancelled = False self._done_callbacks = [] def _block(self, timeout: Optional[float]) -> None: until = None if timeout is None else time.time() + timeout with self._condition: while True: if self._cancelled: raise grpc.FutureCancelledError() elif self._matured: return else: if until is None: self._condition.wait() else: remaining = until - time.time() if remaining < 0: raise grpc.FutureTimeoutError() else: self._condition.wait(timeout=remaining) def _update(self, connectivity: Optional[grpc.ChannelConnectivity]) -> None: with self._condition: if ( not self._cancelled and connectivity is grpc.ChannelConnectivity.READY ): self._matured = True self._channel.unsubscribe(self._update) self._condition.notify_all() done_callbacks = tuple(self._done_callbacks) self._done_callbacks = None else: return for done_callback in done_callbacks: try: done_callback(self) except Exception: # pylint: disable=broad-except _LOGGER.exception(_DONE_CALLBACK_EXCEPTION_LOG_MESSAGE) def cancel(self) -> bool: with self._condition: if not self._matured: self._cancelled = True self._channel.unsubscribe(self._update) self._condition.notify_all() done_callbacks = tuple(self._done_callbacks) self._done_callbacks = None else: return False for done_callback in done_callbacks: try: done_callback(self) except Exception: # pylint: disable=broad-except _LOGGER.exception(_DONE_CALLBACK_EXCEPTION_LOG_MESSAGE) return True def cancelled(self) -> bool: with self._condition: return self._cancelled def running(self) -> bool: with self._condition: return not self._cancelled and not self._matured def done(self) -> bool: with self._condition: return self._cancelled or self._matured def result(self, timeout: Optional[float] = None) -> None: self._block(timeout) def exception(self, timeout: Optional[float] = None) -> None: self._block(timeout) def traceback(self, timeout: Optional[float] = None) -> None: self._block(timeout) def add_done_callback(self, fn: DoneCallbackType): with self._condition: if not self._cancelled and not self._matured: self._done_callbacks.append(fn) return fn(self) def start(self): with self._condition: self._channel.subscribe(self._update, try_to_connect=True) def __del__(self): with self._condition: if not self._cancelled and not self._matured: self._channel.unsubscribe(self._update) def channel_ready_future(channel: grpc.Channel) -> _ChannelReadyFuture: ready_future = _ChannelReadyFuture(channel) ready_future.start() return ready_future def first_version_is_lower(version1: str, version2: str) -> bool: """ Compares two versions in the format '1.60.1' or '1.60.1.dev0'. This method will be used in all stubs generated by grpcio-tools to check whether the stub version is compatible with the runtime grpcio. Args: version1: The first version string. version2: The second version string. Returns: True if version1 is lower, False otherwise. """ version1_list = version1.split(".") version2_list = version2.split(".") try: for i in range(3): if int(version1_list[i]) < int(version2_list[i]): return True elif int(version1_list[i]) > int(version2_list[i]): return False except ValueError: # Return false in case we can't convert version to int. return False # The version without dev0 will be considered lower. return len(version1_list) < len(version2_list)	6.88 KB	2025-05-25 12:26
📁 aio	-	2025-05-25 12:26
📁 beta	-	2025-05-25 12:26
📁 experimental	-	2025-05-25 12:26
📁 framework	-	2025-05-25 12:26

Edit File:

Terminal