ms/py-executor/resource_resolution/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 getLogger, Logger
  16 from types import TracebackType
  17 from typing import Iterable, List, Optional, Type
  18
  19 from grpc import Channel, insecure_channel, secure_channel, ssl_channel_credentials
  20
  21 from blueprints_grpc.proto.BluePrintProcessing_pb2 import ExecutionServiceInput, ExecutionServiceOutput
  22 from blueprints_grpc.proto.BluePrintProcessing_pb2_grpc import BluePrintProcessingServiceStub
  23
  24
  25 class Client:
  26     """Resource resoulution client class."""
  27
  28     def __init__(
  29         self,
  30         server_address: str,
  31         *,
  32         use_ssl: bool = False,
  33         root_certificates: bytes = None,
  34         private_key: bytes = None,
  35         certificate_chain: bytes = None,
  36     ) -> None:
  37         """Client class initialization.
  38
  39         :param server_address: Address to server to connect.
  40         :param use_ssl: Boolean flag to determine if secure channel should be created or not. Keyword argument.
  41         :param root_certificates: The PEM-encoded root certificates. None if it shouldn't be used. Keyword argument.
  42         :param private_key: The PEM-encoded private key as a byte string, or None if no private key should be used. Keyword argument.
  43         :param certificate_chain: The PEM-encoded certificate chain as a byte string to use or or None if no certificate chain should be used. Keyword argument.
  44         """
  45         self.logger = getLogger(__name__)
  46         if use_ssl:
  47             self.channel: Channel = secure_channel(
  48                 server_address, ssl_channel_credentials(root_certificates, private_key, certificate_chain)
  49             )
  50             self.logger.debug(f"Create secure channel to connect with {server_address}")
  51         else:
  52             self.channel: Channel = insecure_channel(server_address)
  53             self.logger.debug(f"Create insecure channel to connect to {server_address}")
  54         self.stub: BluePrintProcessingServiceStub = BluePrintProcessingServiceStub(self.channel)
  55
  56     def close(self) -> None:
  57         """Close client session.
  58
  59         Closes client's channel.
  60         """
  61         self.logger.debug("Close channel connection")
  62         self.channel.close()
  63
  64     def __enter__(self) -> Channel:
  65         """Enter Client instance context.
  66
  67         Return Client instance. In the context user can call methods to communicate with server.
  68         On exit connection with the server is going to be closed.
  69         """
  70         self.logger.debug("Enter Client instance context")
  71         return self
  72
  73     def __exit__(
  74         self,
  75         unused_exc_type: Optional[Type[BaseException]],
  76         unused_exc_value: Optional[BaseException],
  77         unused_traceback: Optional[TracebackType],
  78     ) -> None:
  79         """Exit Client instance context.
  80
  81         Close connection with the server.
  82         """
  83         self.logger.debug("Exit Client instance context")
  84         self.close()
  85
  86     def process(self, messages: Iterable[ExecutionServiceInput]) -> Iterable[ExecutionServiceOutput]:
  87         """Send messages to server and return responses.
  88
  89         :param messages: Iterable messages to send
  90         :return: Iterable responses
  91         """
  92         for message in self.stub.process(messages):
  93             self.logger.debug(f"Get response message: {message}")
  94             yield message