Last active
November 14, 2023 17:24
-
-
Save saintsGrad15/c8eed3a7ba5d3cdbcd3d3783615c7e22 to your computer and use it in GitHub Desktop.
EnvironmentConfig and EnvironmentVariable classes
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
_environment_variables_specs = { | |
**EnvironmentVariable(description="The port at which to expose the reverse proxy on the external interface.", | |
data_type=int, name="PORT", default=8080).as_self_titled_dict(), | |
**EnvironmentVariable(description="The hostname of the DB2 instance.", | |
data_type=str, name="SDG_DB_HOSTNAME", | |
default="db2w-flexule.us-east.db2w.cloud.ibm.com").as_self_titled_dict(), | |
**EnvironmentVariable(description="The DB2 username with which to connect.", | |
data_type=str, name="SDG_DB_USERNAME", required=True).as_self_titled_dict(), | |
**EnvironmentVariable(description="The DB2 user password with which to connect.", | |
data_type=str, name="SDG_DB_PASSWORD", required=True).as_self_titled_dict(), | |
**EnvironmentVariable(description="The name of DB2 schema in which the table is defined.", | |
data_type=str, name="SDG_DB_SCHEMA_NAME", default="TEST_SCHEMA").as_self_titled_dict(), | |
**EnvironmentVariable(description="The name of the DB2 table to which to write.", | |
data_type=str, name="SDG_DB_TABLE_NAME", default="PEOPLE").as_self_titled_dict(), | |
**EnvironmentVariable(description="The level at which SDG should log.", | |
data_type=str, name="SDG_LOG_LEVEL", | |
default=logging.getLevelName(logging.INFO)).as_self_titled_dict() | |
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class EnvironmentConfig: | |
def __init__(self, environment_variables_specs: Dict[str, EnvironmentVariable]): | |
self._environment_variables_specs = environment_variables_specs | |
self.PORT = None # The naming convention is different because this value is automatically injected by Code Engine | |
self.SDG_DB_USERNAME = None | |
self.SDG_DB_PASSWORD = None | |
self.SDG_DB_HOSTNAME = None | |
self.SDG_DB_SCHEMA_NAME = None | |
self.SDG_DB_TABLE_NAME = None | |
self.SDG_LOG_LEVEL = None | |
self.assert_spec_and_config_parity(self._environment_variables_specs) | |
self.print_help_for_cli_flag() | |
self._load() | |
def _load(self): | |
all_environment_variables_valid = True | |
for parameter_name in self.get_attributes_list(): | |
environment_variable: EnvironmentVariable = self._environment_variables_specs[parameter_name] | |
environment_variable.load() | |
if environment_variable.is_valid: | |
setattr(self, parameter_name, environment_variable.value) | |
all_environment_variables_valid = all_environment_variables_valid and environment_variable.is_valid | |
if all_environment_variables_valid is not True: | |
sys.exit(1) | |
return True | |
def print_help_for_cli_flag(self): | |
if "-h" in sys.argv or "--help" in sys.argv: | |
self.print_help() | |
sys.exit(0) | |
def print_help(self): | |
print(""" | |
This application will listen for requests on its "survey" endpoints and write the survey data's contents to a DB2 table. | |
See below for environment variables that dictate its behavior. | |
""") | |
for parameter_name in self.get_attributes_list(): | |
environment_variable: EnvironmentVariable = getattr(self, parameter_name) | |
environment_variable.print_help() | |
def get_attributes_list(self): | |
all_attributes = inspect.getmembers(self) | |
non_callable_attributes = filter(lambda tup: not tup[0].startswith("_") and not callable(tup[1]), | |
all_attributes) | |
non_callable_attribute_names = map(lambda tup: tup[0], non_callable_attributes) | |
return list(non_callable_attribute_names) | |
def assert_spec_and_config_parity(self, environment_variables_specs): | |
environment_variables_class_attribute_names = sorted(self.get_attributes_list()) | |
environment_variables_dict_key_names = sorted(environment_variables_specs.keys()) | |
assert environment_variables_class_attribute_names == environment_variables_dict_key_names, \ | |
f"The data structures EnvironmentVariables and environment_variables_spec don't match. This is a fatal error. Please seek support.\n\n{environment_variables_class_attribute_names}\n{environment_variables_dict_key_names}" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class EnvironmentVariable: | |
def __init__(self, description: str, data_type: Type, name: str, default: Optional[Any] = None, | |
required: Optional[bool] = False, required_if_variable_is_true: Optional[str] = None, | |
validator: Optional[Callable] = None, | |
): | |
self.description = description | |
self.data_type = data_type | |
self.name = name | |
self.default = default | |
self.required = required | |
self.required_if_variable_is_true = required_if_variable_is_true | |
self.validator = validator | |
self._is_valid = None | |
self._value = None | |
@property | |
def is_valid(self): | |
return self._is_valid | |
@property | |
def value(self): | |
return self._value | |
def __repr__(self): | |
return str(self) | |
def __str__(self): | |
return self._value | |
def load(self): | |
variable_string_value = os.environ.get(self.name, self.default) | |
# Validate that required variables are present | |
if self.required is True and variable_string_value is None: | |
logger.error( | |
f"Required environment variable \"{self.name}\" is missing, please set it according to this description: {self.description}") | |
self._is_valid = False | |
required_if_true_variable_name = self.required_if_variable_is_true | |
# Validate dependent variables | |
if required_if_true_variable_name is not None: | |
if os.environ.get(required_if_true_variable_name, "").lower() == "true" and variable_string_value is None: | |
logger.error( | |
f"Environment variable \"{self.name}\" is required if the value of the environment variable \"{required_if_true_variable_name}\" is \"true\" (case insensitive). The latter's value is \"true\" and the former is missing. Please set it according to this description: {self.description}") | |
self._is_valid = False | |
validator = self.validator | |
# Validate against custom logic | |
if validator is not None: | |
validator_error_message = f"The value of the environment variable \"{self.name}\" did not validate. Please set it according to this description: {self.description}" | |
try: | |
valid = validator(variable_string_value) | |
if valid is not True: | |
logger.error(validator_error_message) | |
self._is_valid = False | |
except Exception as e: | |
logger.error(validator_error_message) | |
logger.exception(e) | |
self._is_valid = False | |
variable_type = self.data_type | |
# Coerce variable to its correct type | |
if variable_type is None: | |
self._value = variable_string_value | |
else: | |
try: | |
self._value = variable_type(variable_string_value) | |
except Exception as e: | |
logger.error( | |
f"A failure occurred when the environment variable \"{self.name}\" was being coerced to its type. Please see below the error and set it according to this description: {self.description}") | |
self._is_valid = False | |
self._is_valid = True | |
def print_help(self): | |
message = f"{self.name}: " | |
if self.required is True: | |
message += "(REQUIRED) " | |
message += self.description | |
default_value = self.default | |
if default_value is not None: | |
message += f" Default: {default_value}" | |
print(message) | |
def as_self_titled_dict(self): | |
return { | |
self.name: self | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment