Common - Settings:

# limitations under the License.

import logging, os, time

from typing import List

from typing import Dict, List

from common.Constants import (

    DEFAULT_GRPC_BIND_ADDRESS, DEFAULT_GRPC_GRACE_PERIOD, DEFAULT_GRPC_MAX_WORKERS, DEFAULT_HTTP_BIND_ADDRESS,

    DEFAULT_LOG_LEVEL, DEFAULT_METRICS_PORT, DEFAULT_SERVICE_GRPC_PORTS, DEFAULT_SERVICE_HTTP_BASEURLS,

ENVVAR_SUFIX_SERVICE_PORT_GRPC    = 'SERVICE_PORT_GRPC'

ENVVAR_SUFIX_SERVICE_PORT_HTTP    = 'SERVICE_PORT_HTTP'

def find_missing_environment_variables(

    required_environment_variables : List[str] = []

) -> List[str]:

    if ENVVAR_KUBERNETES_PORT in os.environ:

        missing_variables = set(required_environment_variables).difference(set(os.environ.keys()))

    else:

        # We're not running in Kubernetes, nothing to wait for

        missing_variables = required_environment_variables

    return missing_variables

def find_environment_variables(

    environment_variable_names : List[str] = []

) -> Dict[str, str]:

    environment_variable : Dict[str, str] = dict()

    for name in environment_variable_names:

        if name not in os.environ: continue

        environment_variable[name] = os.environ[name]

    return environment_variable

def wait_for_environment_variables(

    required_environment_variables : List[str] = [], wait_delay_seconds : float = DEFAULT_RESTART_DELAY

):

    missing_variables = find_missing_environment_variables(required_environment_variables)

    if len(missing_variables) == 0: return # We have all environment variables defined

    msg = 'Variables({:s}) are missing in Environment({:s}), restarting in {:f} seconds...'

    LOGGER.error(msg.format(str(missing_variables), str(os.environ), wait_delay_seconds))

) -> None:

    if ENVVAR_KUBERNETES_PORT not in os.environ: return # Not running in Kubernetes

    found = find_environment_variables(required_environment_variables)

    missing = set(required_environment_variables).difference(set(found.keys()))

    if len(missing) == 0: return # We have all environment variables defined

    MSG = 'Variables({:s}) are missing in Environment({:s}), restarting in {:f} seconds...'

    LOGGER.error(MSG.format(str(missing), str(os.environ), wait_delay_seconds))

    time.sleep(wait_delay_seconds)

    raise Exception('Restarting...') # pylint: disable=broad-exception-raised