Skip to content

Instantly share code, notes, and snippets.

@saintsGrad15
Last active November 14, 2023 17:24
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save saintsGrad15/c8eed3a7ba5d3cdbcd3d3783615c7e22 to your computer and use it in GitHub Desktop.
Save saintsGrad15/c8eed3a7ba5d3cdbcd3d3783615c7e22 to your computer and use it in GitHub Desktop.
EnvironmentConfig and EnvironmentVariable classes
_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()
}
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}"
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