Skip to content

Commit

Permalink
add logic for secrets_dir
Browse files Browse the repository at this point in the history
  • Loading branch information
rnag committed Nov 29, 2024
1 parent 80613b6 commit a9027f9
Show file tree
Hide file tree
Showing 4 changed files with 127 additions and 1 deletion.
42 changes: 42 additions & 0 deletions dataclass_wizard/environ/lookups.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import os
from dataclasses import MISSING
from pathlib import Path

from ..decorators import cached_class_property
from ..lazy_imports import dotenv
Expand Down Expand Up @@ -39,6 +40,12 @@ def load_environ(cls, force_reload=False):
# were removed (deleted) from `os.environ`
cls.var_names = set(environ)

if cls._accessed_cleaned_to_env:
cls.cleaned_to_env = {
k: v for k, v in cls.cleaned_to_env.items()
if v in cls.var_names
}

@cached_class_property
def var_names(cls):
"""
Expand Down Expand Up @@ -67,6 +74,41 @@ def reload(cls, env=None):
(clean(var), var) for var in new_vars
)

@classmethod
def secret_values(cls, dirs):
"""
Retrieve the values (environment variables) from secret file(s)
in a secret directory, or a list/tuple of secret directories.
"""
if isinstance(dirs, (str, os.PathLike)):
dirs = [dirs]

env: Environ = {}

for d in dirs:
d: Path = d if isinstance(dirs, os.PathLike) else Path(d)

if d.exists():
if d.is_dir():
# Iterate over all files in the directory
for f in d.iterdir():
if f.is_file(): # Ensure it's a file, not a subdirectory
env[f.name] = f.read_text()
elif d.is_file():
raise ValueError(f'Secrets directory `{d!r}` is a file, not a directory.')

return env

@classmethod
def update_with_secret_values(cls, dirs):

secret_values = cls.secret_values(dirs)

# reload cached mapping of environment variables
cls.reload(secret_values)
# update `environ` with new environment variables
environ.update(secret_values)

@classmethod
def dotenv_values(cls, files):
"""
Expand Down
6 changes: 6 additions & 0 deletions dataclass_wizard/environ/lookups.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,12 @@ class Env:
@classmethod
def reload(cls, env: dict = environ): ...

@classmethod
def secret_values(cls, dirs: EnvFileType) -> Environ: ...

@classmethod
def update_with_secret_values(cls, dirs: EnvFileType): ...

@classmethod
def dotenv_values(cls, files: EnvFileType) -> Environ: ...

Expand Down
13 changes: 12 additions & 1 deletion dataclass_wizard/environ/wizard.py
Original file line number Diff line number Diff line change
Expand Up @@ -150,20 +150,31 @@ def _create_methods(cls):
'Sequence': Sequence,
}

if meta.secrets_dir is None:
_secrets_dir_value = 'None'
else:
_locals['_secrets_dir_value'] = meta.secrets_dir
_secrets_dir_value = '_secrets_dir_value'

# parameters to the `__init__()` method.
init_params = ['self',
'_env_file:EnvFileType=None',
'_reload:bool=False',
f'_env_prefix:str={meta.env_prefix!r}',
'_secrets_dir:EnvFileType|Sequence[EnvFileType]=None',
f'_secrets_dir:EnvFileType|Sequence[EnvFileType]={_secrets_dir_value}',
]

fn_gen = FunctionBuilder()

with fn_gen.function('__init__', init_params, None):

# reload cached var names from `os.environ` as needed.
with fn_gen.if_('_reload'):
fn_gen.add_line('Env.reload()')

with fn_gen.if_('_secrets_dir'):
fn_gen.add_line('Env.update_with_secret_values(_secrets_dir)')

# update environment with values in the "dot env" files as needed.
if _meta_env_file:
fn = fn_gen.elif_
Expand Down
67 changes: 67 additions & 0 deletions tests/unit/environ/test_wizard.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import logging
import os
import tempfile
from dataclasses import field
from datetime import datetime, time, date, timezone
from pathlib import Path
Expand Down Expand Up @@ -537,3 +538,69 @@ class _(EnvWizard.Meta):
int=123)

assert MyPrefixTest() == expected


def test_secrets_dir_and_override():
"""
Test `Meta.secrets_dir` and `_secrets_dir` for handling secrets.
"""
# Create temporary directories and files to simulate secrets
with tempfile.TemporaryDirectory() as default_secrets_dir, tempfile.TemporaryDirectory() as override_secrets_dir:
# Paths for default secrets
default_dir_path = Path(default_secrets_dir)
(default_dir_path / "MY_SECRET_KEY").write_text("default-secret-key")
(default_dir_path / "ANOTHER_SECRET").write_text("default-another-secret")

# Paths for override secrets
override_dir_path = Path(override_secrets_dir)
(override_dir_path / "MY_SECRET_KEY").write_text("override-secret-key")
(override_dir_path / "NEW_SECRET").write_text("new-secret-value")

# Define an EnvWizard class with Meta.secrets_dir
class MySecretClass(EnvWizard):
class _(EnvWizard.Meta):
secrets_dir = default_dir_path # Static default secrets directory

my_secret_key: str
another_secret: str = "default"
new_secret: str = "default-new"

# Test case 1: Use Meta.secrets_dir
instance = MySecretClass()
assert instance.dict() == {
"my_secret_key": "default-secret-key",
"another_secret": "default-another-secret",
"new_secret": "default-new",
}

# Test case 2: Override secrets_dir using _secrets_dir
instance = MySecretClass(_secrets_dir=override_dir_path)
assert instance.dict() == {
"my_secret_key": "override-secret-key", # Overridden by override directory
"another_secret": "default-another-secret", # Still from Meta.secrets_dir
"new_secret": "new-secret-value", # Only in override directory
}

# Test case 3: Missing secrets fallback to defaults
instance = MySecretClass(_reload=True)
assert instance.dict() == {
"my_secret_key": "default-secret-key", # From default directory
"another_secret": "default-another-secret", # From default directory
"new_secret": "default-new", # From the field default
}

# Test case 4: Invalid secrets_dir scenarios
# Case 4a: Directory doesn't exist (ignored with warning)
instance = MySecretClass(_secrets_dir=(default_dir_path, Path("/non/existent/directory")),
_reload=True)
assert instance.dict() == {
"my_secret_key": "default-secret-key", # Fallback to default secrets
"another_secret": "default-another-secret",
"new_secret": "default-new",
}

# Case 4b: secrets_dir is a file (raises error)
with tempfile.NamedTemporaryFile() as temp_file:
invalid_secrets_path = Path(temp_file.name)
with pytest.raises(ValueError, match="Secrets directory .* is a file, not a directory"):
MySecretClass(_secrets_dir=invalid_secrets_path, _reload=True)

0 comments on commit a9027f9

Please sign in to comment.