Skip to content

infrastructure.secrets.env_adapter

src.infrastructure.secrets.env_adapter

Environment variables adapter for local development secrets.

Implements SecretsProtocol using environment variables from .env files.

File: env_adapter.py → class EnvAdapter (PEP 8 naming)

Classes

EnvAdapter

Bases: BaseSecretsAdapter

Local development secrets from .env files.

Converts secret paths to environment variable names
  • 'database/url' → DATABASE_URL
  • 'schwab/api_key' → SCHWAB_API_KEY
Benefits
  • No external dependencies
  • Works offline
  • Fast (no network calls)
  • Familiar to developers (.env files)
Source code in src/infrastructure/secrets/env_adapter.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class EnvAdapter(BaseSecretsAdapter):
    """Local development secrets from .env files.

    Converts secret paths to environment variable names:
        - 'database/url' → DATABASE_URL
        - 'schwab/api_key' → SCHWAB_API_KEY

    Benefits:
        - No external dependencies
        - Works offline
        - Fast (no network calls)
        - Familiar to developers (.env files)
    """

    def __init__(self) -> None:
        """Initialize environment adapter.

        No configuration needed - reads from process environment.
        """
        pass

    def get_secret(self, secret_path: str) -> Result[str, SecretsError]:
        """Get secret from environment variable.

        Args:
            secret_path: Path like 'database/url' or 'schwab/api_key'.

        Returns:
            Success(secret_value) if env var exists.
            Failure(SecretsError) with SECRET_NOT_FOUND if env var not set.

        Example:
            >>> adapter = EnvAdapter()
            >>> result = adapter.get_secret("database/url")
            >>> # If DATABASE_URL env var exists:
            >>> # Success("postgresql://...")
            >>> # If not:
            >>> # Failure(SecretsError(code=SECRET_NOT_FOUND, ...))
        """
        env_var_name = secret_path.replace("/", "_").upper()
        secret_value = os.getenv(env_var_name)

        if secret_value is None:
            return Failure(
                error=SecretsError(
                    code=ErrorCode.SECRET_NOT_FOUND,
                    message=f"Environment variable not found: {env_var_name}",
                    details={"secret_path": secret_path},
                )
            )

        return Success(value=secret_value)

    # get_secret_json() inherited from BaseSecretsAdapter

    def refresh_cache(self) -> None:
        """Clear cache (no-op for environment variables).

        Environment variables are always fresh from the process environment.
        """
        pass
Functions
__init__ 
__init__() -> None

No configuration needed - reads from process environment.

Source code in src/infrastructure/secrets/env_adapter.py 
30
31
32
33
34
35
def __init__(self) -> None:
    """Initialize environment adapter.

    No configuration needed - reads from process environment.
    """
    pass
get_secret 
get_secret(secret_path: str) -> Result[str, SecretsError]

Get secret from environment variable.

Parameters:

Name Type Description Default
secret_path str

Path like 'database/url' or 'schwab/api_key'.

 required

Returns:

Type Description
Result[str, SecretsError]

Success(secret_value) if env var exists.
Result[str, SecretsError]

Failure(SecretsError) with SECRET_NOT_FOUND if env var not set.
Example

adapter = EnvAdapter() result = adapter.get_secret("database/url")

If DATABASE_URL env var exists:
Success("postgresql://...")
If not:
Failure(SecretsError(code=SECRET_NOT_FOUND, ...))
Source code in src/infrastructure/secrets/env_adapter.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def get_secret(self, secret_path: str) -> Result[str, SecretsError]:
    """Get secret from environment variable.

    Args:
        secret_path: Path like 'database/url' or 'schwab/api_key'.

    Returns:
        Success(secret_value) if env var exists.
        Failure(SecretsError) with SECRET_NOT_FOUND if env var not set.

    Example:
        >>> adapter = EnvAdapter()
        >>> result = adapter.get_secret("database/url")
        >>> # If DATABASE_URL env var exists:
        >>> # Success("postgresql://...")
        >>> # If not:
        >>> # Failure(SecretsError(code=SECRET_NOT_FOUND, ...))
    """
    env_var_name = secret_path.replace("/", "_").upper()
    secret_value = os.getenv(env_var_name)

    if secret_value is None:
        return Failure(
            error=SecretsError(
                code=ErrorCode.SECRET_NOT_FOUND,
                message=f"Environment variable not found: {env_var_name}",
                details={"secret_path": secret_path},
            )
        )

    return Success(value=secret_value)
refresh_cache 
refresh_cache() -> None

Clear cache (no-op for environment variables).

Environment variables are always fresh from the process environment.

Source code in src/infrastructure/secrets/env_adapter.py 
71
72
73
74
75
76
def refresh_cache(self) -> None:
    """Clear cache (no-op for environment variables).

    Environment variables are always fresh from the process environment.
    """
    pass