Skip to content

secret

secret

SecretName

Bases: StrEnum

All secret names that can be resolved via read_secret().

Source code in source/src/imarina_load_researchers/core/secret_name.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class SecretName(StrEnum):
    """All secret names that can be resolved via read_secret()."""

    CLIENT_ID = "CLIENT_ID"
    CLIENT_NAME = "CLIENT_NAME"
    CLIENT_SECRET = "CLIENT_SECRET"  # noqa: S105
    DRIVE_ID = "DRIVE_ID"
    SHAREPOINT_DOMAIN = "SHAREPOINT_DOMAIN"
    SITE_NAME = "SITE_NAME"
    TENANT_ID = "TENANT_ID"
    LIST_NAME = "LIST_NAME"

    FTP_HOST = "FTP_HOST"
    FTP_PASSWORD = "FTP_PASSWORD"  # noqa: S105
    FTP_PORT = "FTP_PORT"
    FTP_USER = "FTP_USER"

    SMTP_USERNAME = "SMTP_USERNAME"
    SMTP_PASSWORD = "SMTP_PASSWORD"  # noqa: S105
    SMTP_HOST = "SMTP_HOST"
    SMTP_PORT = "SMTP_PORT"

read_secret 

read_secret(secret_name: SecretName) -> str

Retrieve a secret from predefined sources in order of priority.

Source code in source/src/imarina_load_researchers/core/secret.py 
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
def read_secret(secret_name: SecretName) -> str:
    """Retrieve a secret from predefined sources in order of priority."""
    sources: list[Callable[[], str]] = [
        lambda: read_file_content(f"/run/secrets/{secret_name}"),
        lambda: read_file_content(PROJECT_DIR / "secrets" / secret_name),
        lambda: read_env_var(secret_name),
        lambda: read_vault_secret(secret_name),
    ]

    # Each source signals "not available here" via one of these; anything else
    # (e.g. a programming bug) is left to propagate instead of being swallowed.
    # - read_file_content/read_file: FileNotFoundError, PermissionError (OSError)
    # - read_env_var: KeyError, ValueError
    # - read_vault_secret: KeyError, ValueError, requests.exceptions.RequestException
    expected_errors = (
        OSError,
        KeyError,
        ValueError,
        requests.exceptions.RequestException,
    )
    for source in sources:
        try:
            value = source()
            if value:
                return value
        except expected_errors as e:
            logger.debug(f"Secret source unavailable for '{secret_name}': {e}")
            continue
    raise SecretUnavailableError(secret_name)