Skip to content

domain.protocols.secrets_protocol

src.domain.protocols.secrets_protocol

Secrets management protocol (port) for hexagonal architecture.

This protocol defines what the domain needs from a secrets management system. Infrastructure layer provides concrete implementations (adapters).

Protocol Pattern
  • Domain defines the PORT (this protocol)
  • Infrastructure implements ADAPTERS (EnvAdapter, AWSAdapter, VaultAdapter)
  • Application uses protocol (backend-agnostic)

Classes

SecretsProtocol

Bases: Protocol

Protocol for secrets management systems.

Applications are READ-ONLY consumers of secrets. Secret provisioning is an admin operation (Terraform, AWS CLI, web console).

Implementations
  • EnvAdapter: Local development (.env files)
  • AWSAdapter: Production (AWS Secrets Manager)
  • VaultAdapter: Alternative (HashiCorp Vault)
Source code in src/domain/protocols/secrets_protocol.py 
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
class SecretsProtocol(Protocol):
    """Protocol for secrets management systems.

    Applications are READ-ONLY consumers of secrets.
    Secret provisioning is an admin operation (Terraform, AWS CLI, web console).

    Implementations:
        - EnvAdapter: Local development (.env files)
        - AWSAdapter: Production (AWS Secrets Manager)
        - VaultAdapter: Alternative (HashiCorp Vault)
    """

    def get_secret(self, secret_path: str) -> Result[str, SecretsError]:
        """Get single secret value.

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

        Returns:
            Success(secret_value) if found.
            Failure(SecretsError) if not found or access denied.
        """
        ...

    def get_secret_json(self, secret_path: str) -> Result[dict[str, str], SecretsError]:
        """Get secret as parsed JSON dictionary.

        Args:
            secret_path: Path to JSON-formatted secret.

        Returns:
            Success(parsed_json) if valid JSON.
            Failure(SecretsError) if not found, access denied, or invalid JSON.
        """
        ...

    def refresh_cache(self) -> None:
        """Clear cached secrets to reload after rotation.

        Call this after rotating secrets in backend system.
        Next get_secret() call will fetch fresh value.
        """
        ...
Functions
get_secret 
get_secret(secret_path: str) -> Result[str, SecretsError]

Get single secret value.

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 found.
Result[str, SecretsError]

Failure(SecretsError) if not found or access denied.
Source code in src/domain/protocols/secrets_protocol.py 
30
31
32
33
34
35
36
37
38
39
40
def get_secret(self, secret_path: str) -> Result[str, SecretsError]:
    """Get single secret value.

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

    Returns:
        Success(secret_value) if found.
        Failure(SecretsError) if not found or access denied.
    """
    ...
get_secret_json 
get_secret_json(
    secret_path: str,
) -> Result[dict[str, str], SecretsError]

Get secret as parsed JSON dictionary.

Parameters:

Name Type Description Default
secret_path str

Path to JSON-formatted secret.

 required

Returns:

Type Description
Result[dict[str, str], SecretsError]

Success(parsed_json) if valid JSON.
Result[dict[str, str], SecretsError]

Failure(SecretsError) if not found, access denied, or invalid JSON.
Source code in src/domain/protocols/secrets_protocol.py 
42
43
44
45
46
47
48
49
50
51
52
def get_secret_json(self, secret_path: str) -> Result[dict[str, str], SecretsError]:
    """Get secret as parsed JSON dictionary.

    Args:
        secret_path: Path to JSON-formatted secret.

    Returns:
        Success(parsed_json) if valid JSON.
        Failure(SecretsError) if not found, access denied, or invalid JSON.
    """
    ...
refresh_cache 
refresh_cache() -> None

Clear cached secrets to reload after rotation.

Call this after rotating secrets in backend system. Next get_secret() call will fetch fresh value.

Source code in src/domain/protocols/secrets_protocol.py 
54
55
56
57
58
59
60
def refresh_cache(self) -> None:
    """Clear cached secrets to reload after rotation.

    Call this after rotating secrets in backend system.
    Next get_secret() call will fetch fresh value.
    """
    ...