[rewrite] Added a config manager and some test code
This commit is contained in:
1 parent
807d23da51
commit
3c005dbfa4
10 files changed
+249
-45
No files matched your search
@@ -0,0 +1,3 @@
|
||||
from .config import *
|
||||
|
||||
__all__ = [*config.__all__]
|
||||
@@ -0,0 +1,190 @@
|
||||
import os
|
||||
import logging
|
||||
import re
|
||||
from typing import Any, Type, Optional, Callable
|
||||
|
||||
import dotenv
|
||||
|
||||
from .exceptions import ConfigError
|
||||
|
||||
|
||||
class EnvConfigField():
|
||||
"""
|
||||
Represents a single config field to be parsed from the environment.
|
||||
|
||||
:param env_name: Name of the environment variable
|
||||
:param data_type: Expected data type (int, str, etc.)
|
||||
:param default: Default value if not set in environment
|
||||
:param required: Whether the variable is required
|
||||
:param choices: List of allowed values (for enums)
|
||||
:param min_value: Minimum value (for integers)
|
||||
:param max_value: Maximum value (for integers)
|
||||
:param regex: Regex pattern (for string validation)
|
||||
:param parser: Custom parser function for complex validation
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
env_name: str,
|
||||
data_type: Type[Any] = str,
|
||||
default: Any = None,
|
||||
required: bool = False,
|
||||
choices: Optional[list[Any]] = None,
|
||||
min_value: Optional[int] = None,
|
||||
max_value: Optional[int] = None,
|
||||
regex: Optional[str] = None,
|
||||
parser: Optional[Callable[[str], Any]] = None,
|
||||
):
|
||||
self.value = None
|
||||
self.env_name = env_name
|
||||
self.data_type = data_type
|
||||
self.default = default
|
||||
self.required = required
|
||||
self.choices = choices
|
||||
self.min_value = min_value
|
||||
self.max_value = max_value
|
||||
self.regex = re.compile(regex) if regex else None
|
||||
self.parser = parser
|
||||
|
||||
def load(self):
|
||||
"""Loads the value from the environment and validates it."""
|
||||
raw_value = os.environ.get(self.env_name)
|
||||
|
||||
if self.required and (raw_value is None or raw_value == ""):
|
||||
raise ConfigError(
|
||||
"Field %(env_name)s is required, but was not provided in the environment / .env",
|
||||
env_name=self.env_name,
|
||||
)
|
||||
|
||||
self.value = self.validate(raw_value)
|
||||
|
||||
def validate(self, value: Any) -> Any:
|
||||
"""Validates the value based on provided constraints."""
|
||||
|
||||
# Use the default value if available and skip validation
|
||||
if (value is None or value == "") and self.default is not None:
|
||||
return self.default
|
||||
|
||||
# Use custom parser if provided
|
||||
if self.parser:
|
||||
value = self.parser(value)
|
||||
|
||||
# Convert value to expected data type
|
||||
if self.data_type == str:
|
||||
value = self.__parse_to_str(value)
|
||||
elif self.data_type == int:
|
||||
value = self.__parse_to_int(value)
|
||||
elif self.data_type == bool:
|
||||
value = self.__parse_to_bool(value)
|
||||
|
||||
# Validate allowed choices
|
||||
if self.choices and value not in self.choices:
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)s is not one of the following allowed values: %(allowed_values)s",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
allowed_values=", ".join(map(str, self.choices)),
|
||||
)
|
||||
|
||||
return value
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
return self.value
|
||||
|
||||
@value.setter
|
||||
def value(self, new):
|
||||
raise TypeError("Cannot modify")
|
||||
|
||||
def __parse_to_str(self, value) -> str:
|
||||
"""Parses and validates string values."""
|
||||
value = str(value)
|
||||
|
||||
if self.regex and not self.regex.match(value):
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)s does not match the expected pattern: %(regex)s",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
regex=self.regex.pattern,
|
||||
)
|
||||
|
||||
return value
|
||||
|
||||
def __parse_to_int(self, value) -> int:
|
||||
"""Parses and validates integer values."""
|
||||
if not str(value).isdigit():
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)s must be a numeric integer",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
)
|
||||
|
||||
value = int(value)
|
||||
|
||||
if self.min_value is not None and value < self.min_value:
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)d is too small. Must be greater than or equal to %(min_value)d",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
min_value=self.min_value,
|
||||
)
|
||||
|
||||
if self.max_value is not None and value > self.max_value:
|
||||
raise ConfigError(
|
||||
"Value for %(env_name)s - %(value)d is too large. Must be smaller than or equal to %(max_value)d",
|
||||
env_name=self.env_name,
|
||||
value=value,
|
||||
max_value=self.max_value,
|
||||
)
|
||||
|
||||
return value
|
||||
|
||||
def __parse_to_bool(self, value):
|
||||
value = str(value).lower()
|
||||
|
||||
true_values = ["1", "true", "yes"]
|
||||
false_values = ["0", "false", "no"]
|
||||
|
||||
if value in true_values:
|
||||
return True
|
||||
if value in false_values:
|
||||
return False
|
||||
|
||||
raise ConfigError("Value for %(env_name)s - %(value)s cannot be converted to a true / false value",
|
||||
env_name=self.env_name,
|
||||
value=value)
|
||||
|
||||
|
||||
class EnvConfig():
|
||||
def __init__(self):
|
||||
self.logger = logging.getLogger(__name__)
|
||||
|
||||
self.fields = {}
|
||||
|
||||
def add_field(self, name: str, field: EnvConfigField):
|
||||
if name in self.fields:
|
||||
return
|
||||
|
||||
self.fields[name] = field
|
||||
|
||||
def load_config(self):
|
||||
dotenv.load_dotenv()
|
||||
|
||||
for field in self.fields.values():
|
||||
try:
|
||||
field.load
|
||||
except ConfigError as e:
|
||||
self.logger.error("Configuration error: %s", e)
|
||||
raise
|
||||
|
||||
def __getitem__(self, key: str):
|
||||
return self.fields["key"].value
|
||||
|
||||
def __contains__(self, key: str) -> bool:
|
||||
return key in self.fields
|
||||
|
||||
def to_dict(self):
|
||||
return {attr: getattr(self, attr) for attr in self.fields}
|
||||
|
||||
|
||||
__all__ = ["EnvConfig"]
|
||||
@@ -0,0 +1,12 @@
|
||||
class SwagShopError(Exception):
|
||||
def __init__(self, message):
|
||||
super().__init__(message)
|
||||
self.message = message
|
||||
|
||||
class ConfigError(SwagShopError):
|
||||
def __init__(self, message, **kwargs):
|
||||
formatted_message = message % kwargs
|
||||
super().__init__(formatted_message)
|
||||
self.message = formatted_message
|
||||
|
||||
__all__ = ["SwagShopError", "ConfigError"]
|
||||
Reference in new issue
Block a user