From 88da6c31d28cb1f69ade02b021c1de1eb9689c33 Mon Sep 17 00:00:00 2001 From: Tiago Requeijo Date: Tue, 23 Jul 2024 08:07:02 -0400 Subject: [PATCH] Feature/strip prefixes (#102) * Add a new paramter to strip the prefixes * Strip prefixes, and fixed TOML tests --- CHANGELOG.md | 16 ++++++++ pyproject.toml | 3 +- src/config/__init__.py | 69 ++++++++++++++++++++++++++------- src/config/contrib/gcp.py | 2 +- tests/contrib/test_azure.py | 4 +- tests/test_configuration_set.py | 10 ++++- tests/test_dotenv.py | 14 ++++++- tests/test_env.py | 16 ++++++++ tests/test_nested.py | 30 ++++++++++++++ tests/test_toml.py | 39 ++++++++++++++++--- 10 files changed, 178 insertions(+), 25 deletions(-) create mode 100644 tests/test_nested.py diff --git a/CHANGELOG.md b/CHANGELOG.md index cb2b4bc..c862554 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,22 @@ All notable changes to this project will be documented in this file. ## [Unreleased] + +## [0.12.0] - 2024-07-23 + +### Added + +- Granular `strip_prefix` parameters across different config types + +### Fixed + +- Unit tests for .toml files + +### Changed + +- Enviroment files are now loaded from filenames with a suffix of `.env` or starting with `.env` + + ## [0.11.0] - 2024-04-23 ### Changed diff --git a/pyproject.toml b/pyproject.toml index 4cea822..abe62a3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,7 +28,7 @@ Source = "https://github.com/tr11/python-configuration" [project.optional-dependencies] # cloud aws = ["boto3>=1.28.20"] -azure = ["azure-keyvault>=5.0.0"] +azure = ["azure-keyvault>=4.2.0", "azure-identity"] gcp = ["google-cloud-secret-manager>=2.16.3"] vault = ["hvac>=1.1.1"] # file formats @@ -148,6 +148,7 @@ module = [ 'hvac.exceptions', 'jsonschema', 'jsonschema.exceptions', + 'azure.identity', ] ignore_missing_imports = true diff --git a/src/config/__init__.py b/src/config/__init__.py index f50d33b..34236e3 100644 --- a/src/config/__init__.py +++ b/src/config/__init__.py @@ -32,6 +32,7 @@ def config( *configs: Iterable, prefix: str = "", + strip_prefix: bool = True, separator: Optional[str] = None, remove_level: int = 1, lowercase_keys: bool = False, @@ -44,6 +45,7 @@ def config( Params: configs: iterable of configurations prefix: prefix to filter environment variables with + strip_prefix: whether to strip the prefix remove_level: how many levels to remove from the resulting config lowercase_keys: whether to convert every key to lower case. ignore_missing_paths: whether to ignore failures from missing files/folders. @@ -71,6 +73,9 @@ def config( if isinstance(config_, Mapping): instances.append(config_from_dict(config_, **default_kwargs)) continue + elif isinstance(config_, Configuration): + instances.append(config_) + continue elif isinstance(config_, str): if config_.endswith(".py"): config_ = ("python", config_, *default_args) @@ -82,8 +87,8 @@ def config( config_ = ("toml", config_, True) elif config_.endswith(".ini"): config_ = ("ini", config_, True) - elif config_.endswith(".env"): - config_ = ("dotenv", config_, True) + elif config_.endswith(".env") or config_.startswith(".env"): + config_ = ("dotenv", config_, True, *default_args) elif os.path.isdir(config_): config_ = ("path", config_, remove_level) elif config_ in ("env", "environment"): @@ -103,7 +108,13 @@ def config( instances.append(config_from_dict(*config_[1:], **default_kwargs)) elif type_ in ("env", "environment"): params = list(config_[1:]) + default_args[(len(config_) - 1) :] - instances.append(config_from_env(*params, **default_kwargs)) + instances.append( + config_from_env( + *params, + **default_kwargs, + strip_prefix=strip_prefix, + ), + ) elif type_ == "python": if len(config_) < 2: raise ValueError("No path specified for python module") @@ -113,6 +124,7 @@ def config( *params, **default_kwargs, ignore_missing_paths=ignore_missing_paths, + strip_prefix=strip_prefix, ), ) elif type_ == "json": @@ -137,6 +149,7 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, + strip_prefix=strip_prefix, ), ) elif type_ == "ini": @@ -145,6 +158,7 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, + strip_prefix=strip_prefix, ), ) elif type_ == "dotenv": @@ -153,6 +167,7 @@ def config( *config_[1:], **default_kwargs, ignore_missing_paths=ignore_missing_paths, + strip_prefix=strip_prefix, ), ) elif type_ == "path": @@ -178,9 +193,10 @@ class EnvConfiguration(Configuration): def __init__( self, - prefix: str, + prefix: str = "", separator: str = "__", *, + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -189,9 +205,11 @@ def __init__( prefix: prefix to filter environment variables with separator: separator to replace by dots + strip_prefix: whether to include the prefix lowercase_keys: whether to convert every key to lower case. """ self._prefix = prefix + self._strip_prefix = strip_prefix self._separator = separator super().__init__( {}, @@ -207,9 +225,12 @@ def reload(self) -> None: for key, value in os.environ.items(): if not key.startswith(self._prefix + self._separator): continue - result[ - key[len(self._prefix) :].replace(self._separator, ".").strip(".") - ] = value + if self._strip_prefix: + result[ + key[len(self._prefix) :].replace(self._separator, ".").strip(".") + ] = value + else: + result[key.replace(self._separator, ".").strip(".")] = value super().__init__( result, lowercase_keys=self._lowercase, @@ -222,6 +243,7 @@ def config_from_env( prefix: str, separator: str = "__", *, + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -231,6 +253,7 @@ def config_from_env( Params: prefix: prefix to filter environment variables with. separator: separator to replace by dots. + strip_prefix: whether to include the prefix lowercase_keys: whether to convert every key to lower case. interpolate: whether to apply string interpolation when looking for items. @@ -240,6 +263,7 @@ def config_from_env( return EnvConfiguration( prefix, separator, + strip_prefix=strip_prefix, lowercase_keys=lowercase_keys, interpolate=interpolate, interpolate_type=interpolate_type, @@ -463,6 +487,7 @@ def __init__( read_from_file: bool = False, *, section_prefix: str = "", + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -470,6 +495,7 @@ def __init__( ): """Class Constructor.""" self._section_prefix = section_prefix + self._strip_prefix = strip_prefix super().__init__( data=data, read_from_file=read_from_file, @@ -502,8 +528,9 @@ def optionxform(self, optionstr: str) -> str: data = cast(str, data) cfg = ConfigParser() cfg.read_string(data) + n = len(self._section_prefix) if self._strip_prefix else 0 result = { - section[len(self._section_prefix) :] + "." + k: v + section[n:] + "." + k: v for section, values in cfg.items() for k, v in values.items() if section.startswith(self._section_prefix) @@ -516,6 +543,7 @@ def config_from_ini( read_from_file: bool = False, *, section_prefix: str = "", + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -538,6 +566,7 @@ def config_from_ini( data, read_from_file, section_prefix=section_prefix, + strip_prefix=strip_prefix, lowercase_keys=lowercase_keys, interpolate=interpolate, interpolate_type=interpolate_type, @@ -555,6 +584,7 @@ def __init__( prefix: str = "", separator: str = "__", *, + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -562,7 +592,9 @@ def __init__( ): """Class Constructor.""" self._prefix = prefix + self._strip_prefix = strip_prefix self._separator = separator + super().__init__( data=data, read_from_file=read_from_file, @@ -589,8 +621,9 @@ def _reload( parse_env_line(x) for x in data.splitlines() if x and not x.startswith("#") ) + n = len(self._prefix) if self._strip_prefix else 0 result = { - k[len(self._prefix) :].replace(self._separator, ".").strip("."): v + k[n:].replace(self._separator, ".").strip("."): v for k, v in result.items() if k.startswith(self._prefix) } @@ -604,6 +637,7 @@ def config_from_dotenv( prefix: str = "", separator: str = "__", *, + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -629,6 +663,7 @@ def config_from_dotenv( read_from_file, prefix=prefix, separator=separator, + strip_prefix=strip_prefix, lowercase_keys=lowercase_keys, interpolate=interpolate, interpolate_type=interpolate_type, @@ -645,6 +680,7 @@ def __init__( prefix: str = "", separator: str = "_", *, + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -677,6 +713,7 @@ def __init__( module = importlib.import_module(module) self._module: Optional[ModuleType] = module self._prefix = prefix + self._strip_prefix = strip_prefix self._separator = separator except (FileNotFoundError, ModuleNotFoundError): if not ignore_missing_paths: @@ -699,10 +736,9 @@ def reload(self) -> None: for x in dir(self._module) if not x.startswith("__") and x.startswith(self._prefix) ] + n = len(self._prefix) if self._strip_prefix else 0 result = { - k[len(self._prefix) :] - .replace(self._separator, ".") - .strip("."): getattr(self._module, k) + k[n:].replace(self._separator, ".").strip("."): getattr(self._module, k) for k in variables } else: @@ -720,6 +756,7 @@ def config_from_python( prefix: str = "", separator: str = "_", *, + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -741,6 +778,7 @@ def config_from_python( module, prefix, separator, + strip_prefix=strip_prefix, lowercase_keys=lowercase_keys, interpolate=interpolate, interpolate_type=interpolate_type, @@ -882,6 +920,7 @@ def __init__( read_from_file: bool = False, *, section_prefix: str = "", + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -894,6 +933,7 @@ def __init__( ) self._section_prefix = section_prefix + self._strip_prefix = strip_prefix super().__init__( data=data, read_from_file=read_from_file, @@ -920,8 +960,9 @@ def _reload( loaded = toml.loads(data) loaded = cast(dict, loaded) + n = len(self._section_prefix) if self._section_prefix else 0 result = { - k[len(self._section_prefix) :]: v + k[n:]: v for k, v in self._flatten_dict(loaded).items() if k.startswith(self._section_prefix) } @@ -934,6 +975,7 @@ def config_from_toml( read_from_file: bool = False, *, section_prefix: str = "", + strip_prefix: bool = True, lowercase_keys: bool = False, interpolate: InterpolateType = False, interpolate_type: InterpolateEnumType = InterpolateEnumType.STANDARD, @@ -955,6 +997,7 @@ def config_from_toml( data, read_from_file, section_prefix=section_prefix, + strip_prefix=strip_prefix, lowercase_keys=lowercase_keys, interpolate=interpolate, interpolate_type=interpolate_type, diff --git a/src/config/contrib/gcp.py b/src/config/contrib/gcp.py index d346b3a..a7865c3 100644 --- a/src/config/contrib/gcp.py +++ b/src/config/contrib/gcp.py @@ -35,7 +35,7 @@ class GCPSecretManagerConfiguration(Configuration): def __init__( self, project_id: str, - credentials: Credentials = None, + credentials: Optional[Credentials] = None, client_options: Optional[ClientOptions] = None, cache_expiration: int = 5 * 60, interpolate: InterpolateType = False, diff --git a/tests/contrib/test_azure.py b/tests/contrib/test_azure.py index 1554e35..1524b14 100644 --- a/tests/contrib/test_azure.py +++ b/tests/contrib/test_azure.py @@ -5,12 +5,12 @@ from config import config_from_dict try: - import azure from config.contrib.azure import AzureKeyVaultConfiguration from azure.core.exceptions import ResourceNotFoundError + azure = True except ImportError: # pragma: no cover azure = None # type: ignore - + raise DICT = { "foo": "foo_val", diff --git a/tests/test_configuration_set.py b/tests/test_configuration_set.py index 2886b7b..482a6c9 100644 --- a/tests/test_configuration_set.py +++ b/tests/test_configuration_set.py @@ -154,6 +154,12 @@ dotenv3 = xyz """ +DOTENV_PREFIX = """ +CONFIG__dotenv1 = abc +CONFIG__dotenv2 = 1.2 +CONFIG__dotenv3 = xyz +""" + DICT_DOTENV = { "dotenv1": "abc", @@ -437,7 +443,7 @@ def test_alternate_set_loader_strings(): # type: ignore f2.file.write(INI.encode()) f2.file.flush() # ini - f5.file.write(DOTENV.encode()) + f5.file.write(DOTENV_PREFIX.encode()) f5.file.flush() entries = [ @@ -804,7 +810,7 @@ def test_reload(): # type: ignore f2.file.write(INI.encode()) f2.file.flush() # ini - f5.file.write(DOTENV.encode()) + f5.file.write(DOTENV_PREFIX.encode()) f5.file.flush() entries = [ diff --git a/tests/test_dotenv.py b/tests/test_dotenv.py index d492914..a6879dc 100644 --- a/tests/test_dotenv.py +++ b/tests/test_dotenv.py @@ -1,5 +1,5 @@ import pytest -from config import config_from_dotenv, config_from_dict +from config import config, config_from_dotenv, config_from_dict import tempfile DOTENV = """ @@ -68,6 +68,18 @@ def test_load_dotenv_filename(): # type: ignore assert cfg == config_from_dict(dict((k, str(v)) for k, v in DICT.items())) +def test_load_dotenv_config(): # type: ignore + with tempfile.NamedTemporaryFile(suffix='.env') as f: + f.file.write(DOTENV_WITH_PREFIXES.encode()) + f.file.flush() + cfg = config(f.name, lowercase_keys=True, prefix="PREFIX") + + print(cfg) + print(config_from_dict(DICT_WITH_PREFIXES)) + assert cfg == config_from_dict(DICT_WITH_PREFIXES) + + + def test_reload(): # type: ignore with tempfile.NamedTemporaryFile() as f: f.file.write(DOTENV.encode()) diff --git a/tests/test_env.py b/tests/test_env.py index ef13377..cd06e5b 100644 --- a/tests/test_env.py +++ b/tests/test_env.py @@ -55,3 +55,19 @@ def test_reload(): # type: ignore d = DICT.copy() d["a2.b2.c3"] = "updated" assert cfg == config_from_dict(dict((k, str(v)) for k, v in d.items())) + + +def test_reload_2(): # type: ignore + os.environ.update( + (PREFIX + "__" + k.replace(".", "__").upper(), str(v)) for k, v in DICT.items() + ) + + cfg = config_from_env(PREFIX, lowercase_keys=True, strip_prefix=False) + assert cfg == config_from_dict(dict((PREFIX.lower() + "." + k, str(v)) for k, v in DICT.items())) + + os.environ[PREFIX + "__" + "A2__B2__C3"] = "updated" + assert cfg == config_from_dict(dict((PREFIX.lower() + "." + k, str(v)) for k, v in DICT.items())) + cfg.reload() + d = DICT.copy() + d["a2.b2.c3"] = "updated" + assert cfg == config_from_dict(dict((PREFIX.lower() + "." + k, str(v)) for k, v in d.items())) diff --git a/tests/test_nested.py b/tests/test_nested.py new file mode 100644 index 0000000..81db9a5 --- /dev/null +++ b/tests/test_nested.py @@ -0,0 +1,30 @@ +from config import config_from_dict, config + + +DICT = { + "a1.B1.c1": 1, + "a1.b1.C2": 2, + "A1.b1.c3": 3, + "a1.b2.c1": "a", + "a1.b2.c2": True, + "a1.b2.c3": 1.1, + "a2.b1.c1": "f", + "a2.b1.c2": False, + "a2.b1.c3": None, + "a2.b2.c1": 10, + "a2.b2.c2": "YWJjZGVmZ2g=", + "a2.b2.c3": "abcdefgh", +} + +NESTED = { + "a1": {"b1": {"c1": 1, "C2": 2, "c3": 3}, "b2": {"c1": "a", "c2": True, "c3": 1.1}} +} + + +def test_load_from_config(): # type: ignore + cfg1 = config_from_dict(DICT, lowercase_keys=True) + cfg2 = config_from_dict(NESTED, lowercase_keys=True) + + assert config(DICT, NESTED, lowercase_keys=True ) == config(cfg1, NESTED, lowercase_keys=True ) + assert config(DICT, NESTED, lowercase_keys=True ) == config(DICT, cfg2, lowercase_keys=True ) + assert config(DICT, NESTED, lowercase_keys=True ) == config(cfg1, cfg2, lowercase_keys=True ) diff --git a/tests/test_toml.py b/tests/test_toml.py index 665587f..46cddcc 100644 --- a/tests/test_toml.py +++ b/tests/test_toml.py @@ -1,16 +1,20 @@ from pathlib import Path +import sys import tempfile import pytest -from config import config_from_dict +from config import config_from_dict, config try: - import toml - + if sys.version_info < (3, 11): # pragma: no cover + import tomli as toml + else: # pragma: no cover + import tomllib as toml + from config import config_from_toml except ImportError: - toml = None + toml = None # type: ignore config_from_toml = None # type: ignore @@ -30,7 +34,20 @@ } if toml: - TOML = toml.dumps(DICT) + TOML = """ +"a1.b1.c1" = 1 +"a1.b1.c2" = 2 +"a1.b1.c3" = 3 +"a1.b2.c1" = "a" +"a1.b2.c2" = true +"a1.b2.c3" = 1.1 +"a2.b1.c1" = "f" +"a2.b1.c2" = false +"a2.b1.c3" = "" +"a2.b2.c1" = 10 +"a2.b2.c2" = "YWJjZGVmZ2g=" +"a2.b2.c3" = "abcdefgh" +""" TOML2 = """ [owner] @@ -110,6 +127,18 @@ def test_load_toml_filename_2(): # type: ignore assert cfg == config_from_dict(DICT) +@pytest.mark.skipif("toml is None") +def test_load_toml_filename_3(): # type: ignore + with tempfile.NamedTemporaryFile(suffix='.toml') as f: + f.file.write(TOML.encode()) + f.file.flush() + cfg = config(f.name) + assert cfg["a1.b1.c1"] == 1 + assert cfg["a1.b1"].as_dict() == {"c1": 1, "c2": 2, "c3": 3} + assert cfg["a1.b2"].as_dict() == {"c1": "a", "c2": True, "c3": 1.1} + assert cfg == config_from_dict(DICT) + + @pytest.mark.skipif("toml is None") def test_equality(): # type: ignore cfg = config_from_toml(TOML)