ms/py-executor/resource_resolution/grpc/client.py

   1 """Copyright 2019 Deutsche Telekom.
   2
   3 Licensed under the Apache License, Version 2.0 (the "License");
   4 you may not use this file except in compliance with the License.
   5 You may obtain a copy of the License at
   6
   7     http://www.apache.org/licenses/LICENSE-2.0
   8
   9 Unless required by applicable law or agreed to in writing, software
  10 distributed under the License is distributed on an "AS IS" BASIS,
  11 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 See the License for the specific language governing permissions and
  13 limitations under the License.
  14 """
  15 from logging import Logger, getLogger
  16 from types import TracebackType
  17 from typing import Iterable, Optional, Type
  18
  19 from grpc import (
  20     Channel,
  21     insecure_channel,
  22     intercept_channel,
  23     secure_channel,
  24     ssl_channel_credentials,
  25 )
  26 from proto.BluePrintProcessing_pb2 import ExecutionServiceInput, ExecutionServiceOutput
  27 from proto.BluePrintProcessing_pb2_grpc import BluePrintProcessingServiceStub
  28
  29 from .authorization import AuthTokenInterceptor
  30
  31
  32 class Client:
  33     """Resource resoulution client class."""
  34
  35     def __init__(
  36         self,
  37         server_address: str,
  38         *,
  39         # TLS/SSL configuration
  40         use_ssl: bool = False,
  41         root_certificates: bytes = None,
  42         private_key: bytes = None,
  43         certificate_chain: bytes = None,
  44         # Authentication header configuration
  45         use_header_auth: bool = False,
  46         header_auth_token: str = None,
  47     ) -> None:
  48         """Client class initialization.
  49
  50         :param server_address: Address to server to connect.
  51         :param use_ssl: Boolean flag to determine if secure channel should be created or not. Keyword argument.
  52         :param root_certificates: The PEM-encoded root certificates. None if it shouldn't be used. Keyword argument.
  53         :param private_key: The PEM-encoded private key as a byte string, or None if no private key should be used.
  54             Keyword argument.
  55         :param certificate_chain: The PEM-encoded certificate chain as a byte string to use or or None if
  56             no certificate chain should be used. Keyword argument.
  57         :param use_header_auth: Boolean flag to determine if authorization headed shoud be added for every call or not.
  58             Keyword argument.
  59         :param header_auth_token: Authorization token value. Keyword argument.
  60         """
  61         self.logger: Logger = getLogger(__name__)
  62         if use_ssl:
  63             self.channel: Channel = secure_channel(
  64                 server_address, ssl_channel_credentials(root_certificates, private_key, certificate_chain)
  65             )
  66             self.logger.debug(f"Create secure channel to connect with {server_address}")
  67         else:
  68             self.channel: Channel = insecure_channel(server_address)
  69             self.logger.debug(f"Create insecure channel to connect to {server_address}")
  70         if use_header_auth:
  71             self.channel: Channel = intercept_channel(self.channel, AuthTokenInterceptor(header_auth_token))
  72         self.stub: BluePrintProcessingServiceStub = BluePrintProcessingServiceStub(self.channel)
  73
  74     def close(self) -> None:
  75         """Close client session.
  76
  77         Closes client's channel.
  78         """
  79         self.logger.debug("Close channel connection")
  80         self.channel.close()
  81
  82     def __enter__(self) -> Channel:
  83         """Enter Client instance context.
  84
  85         Return Client instance. In the context user can call methods to communicate with server.
  86         On exit connection with the server is going to be closed.
  87         """
  88         self.logger.debug("Enter Client instance context")
  89         return self
  90
  91     def __exit__(
  92         self,
  93         unused_exc_type: Optional[Type[BaseException]],
  94         unused_exc_value: Optional[BaseException],
  95         unused_traceback: Optional[TracebackType],
  96     ) -> None:
  97         """Exit Client instance context.
  98
  99         Close connection with the server.
 100         """
 101         self.logger.debug("Exit Client instance context")
 102         self.close()
 103
 104     def process(self, messages: Iterable[ExecutionServiceInput]) -> Iterable[ExecutionServiceOutput]:
 105         """Send messages to server and return responses.
 106
 107         :param messages: Iterable messages to send
 108         :return: Iterable responses
 109         """
 110         for message in self.stub.process(messages):
 111             self.logger.debug(f"Get response message: {message}")
 112             yield message