Skip to content

deep.grpc.grpc_service

Service for connecting to GRPC channel.

GRPCService

This service handles config and initialising the GRPc channel that will be used.

Source code in deep/grpc/grpc_service.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class GRPCService:
    """This service handles config and initialising the GRPc channel that will be used."""

    def __init__(self, config: ConfigService):
        """
        Create a new grpc service.

        :param config: the deep config
        """
        self.channel = None
        self._config = config
        self._service_url = config.SERVICE_URL
        self._secure = config.SERVICE_SECURE
        self._metadata = None

    def start(self):
        """Start and connect the GRPC channel."""
        if str2bool(self._secure):
            logging.info("Connecting securely")
            logging.debug("Connecting securely to: %s", self._service_url)
            self.channel = grpc.secure_channel(self._service_url, grpc.ssl_channel_credentials())
        else:
            logging.info("Connecting with insecure channel")
            logging.debug("Connecting with insecure channel to: %s ", self._service_url)
            self.channel = grpc.insecure_channel(self._service_url)

    def metadata(self):
        """
        Get GRPC metadata.

        Call this to get any metadata that should be attached to calls.

        :return: list of metadata
        """
        if self._metadata is None:
            self._metadata = self._build_metadata()
        return self._metadata

    def _build_metadata(self):
        provider = AuthProvider.get_provider(self._config)
        if provider is not None:
            return provider.provide()
        return []

__init__(config)

Create a new grpc service.

:param config: the deep config

Source code in deep/grpc/grpc_service.py 
29
30
31
32
33
34
35
36
37
38
39
def __init__(self, config: ConfigService):
    """
    Create a new grpc service.

    :param config: the deep config
    """
    self.channel = None
    self._config = config
    self._service_url = config.SERVICE_URL
    self._secure = config.SERVICE_SECURE
    self._metadata = None

metadata()

Get GRPC metadata.

Call this to get any metadata that should be attached to calls.

:return: list of metadata

Source code in deep/grpc/grpc_service.py 
52
53
54
55
56
57
58
59
60
61
62
def metadata(self):
    """
    Get GRPC metadata.

    Call this to get any metadata that should be attached to calls.

    :return: list of metadata
    """
    if self._metadata is None:
        self._metadata = self._build_metadata()
    return self._metadata

start()

Start and connect the GRPC channel.

Source code in deep/grpc/grpc_service.py 
41
42
43
44
45
46
47
48
49
50
def start(self):
    """Start and connect the GRPC channel."""
    if str2bool(self._secure):
        logging.info("Connecting securely")
        logging.debug("Connecting securely to: %s", self._service_url)
        self.channel = grpc.secure_channel(self._service_url, grpc.ssl_channel_credentials())
    else:
        logging.info("Connecting with insecure channel")
        logging.debug("Connecting with insecure channel to: %s ", self._service_url)
        self.channel = grpc.insecure_channel(self._service_url)