Skip to content

Instantly share code, notes, and snippets.

@keegoo
Last active December 4, 2023 13:30
Show Gist options
  • Save keegoo/ec90fe4d257fa500b3b60c4ef2e6dbb2 to your computer and use it in GitHub Desktop.
Save keegoo/ec90fe4d257fa500b3b60c4ef2e6dbb2 to your computer and use it in GitHub Desktop.
This is the code example for blog - An implementation of Config class
import os
import yaml
from mergedeep import merge
from functools import reduce
class Config:
class ValueDesc:
def __init__(self, env_var, type=str, required=True, default=None):
self.env_var = env_var
self.type = type
self.required = required
self.default = default
PARAMETERS = {
"web.request.timeout": ValueDesc("WEB_MAX_REQUEST_TIMEOUT", type=int, required=False, default=600),
"web.request.maxSize": ValueDesc("WEB_MAX_HEADER_SIZE", type=int, required=False, default=32768),
"web.headers.clientType": ValueDesc("WEB_HEADERS_CLIENT_TYPE", type=str, required=True, default="Customize-Client"),
"web.headers.maxSize": ValueDesc("WEB_HEADERS_MAX_SIZE", type=int, required=True, default=1024),
"database.username": ValueDesc("DB_USERNAME", type=str, required=True),
"database.password": ValueDesc("DB_PASSWORD", type=str, required=True),
"database.schema": ValueDesc("DB_SCHEMA", type=str, required=True),
"database.poolSize": ValueDesc("DB_POOL_SIZE", type=int, required=False, default=10)
}
def __init__(self):
self.web_request_timeout = self._get_config_value("web.request.timeout")
self.web_request_maxSize = self._get_config_value("web.request.maxSize")
self.web_headers_clientType = self._get_config_value("web.headers.clientType")
self.web_headers_maxSize = self._get_config_value("web.headers.maxSize")
self.database_username = self._get_config_value("database.username")
self.database_password = self._get_config_value("database.password")
self.database_schema = self._get_config_value("database.schema")
self.database_poolSize = self._get_config_value("database.poolSize")
def generate_yaml_with_default_values(self):
res = []
for key, value in Config.PARAMETERS.items():
items = key.split(".")[::-1]
default = value.default
x = reduce(lambda v, k: {k: default} if v is None else {k: v}, items, None)
res.append(x)
x = reduce(merge, res)
return yaml.dump(x, default_flow_style=False)
def _get_config_value(self, key):
config_value = yaml.safe_load(open("config.yaml", "r").read())
value = reduce(lambda data, k: data[k] if data and k in data else None, key.split('.'), config_value)
value_desc = Config.PARAMETERS.get(key)
if value is None:
if not value_desc:
return None
value = os.getenv(value_desc.env_var, value_desc.default)
return value
# config = Config()
# print(config.generate_yaml_with_default_values())
# print(config.web_request_timeout)
# print(config.web_request_maxSize)
# print(config.web_headers_clientType)
# print(config.web_headers_maxSize)
# print(config.database_username)
# print(config.database_password)
# print(config.database_schema)
# print(config.database_poolSize)
web:
request:
timeout: 1
maxSize: 2
headers:
clientType: 'custom'
maxSize: 3
database:
username: "admin"
password: "password"
schema: "default"
poolSize: 4
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment