airbyte.secrets.env_vars

 1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 2"""Secret manager that retrieves secrets from environment variables and `.env` files."""
 3
 4from __future__ import annotations
 5
 6import os
 7from typing import TYPE_CHECKING
 8
 9from dotenv import dotenv_values
10
11from airbyte.secrets.base import SecretManager, SecretSourceEnum, SecretString
12
13
14if TYPE_CHECKING:
15    from pathlib import Path
16
17
18class EnvVarSecretManager(SecretManager):
19    """Secret manager that retrieves secrets from environment variables."""
20
21    name = SecretSourceEnum.ENV.value
22
23    def get_secret(self, secret_name: str) -> SecretString | None:
24        """Get a named secret from the environment."""
25        if secret_name not in os.environ:
26            return None
27
28        return SecretString(os.environ[secret_name])
29
30
31class DotenvSecretManager(SecretManager):
32    """Secret manager that retrieves secrets from a `.env` file."""
33
34    dotenv_path: Path | None = None
35
36    @property
37    def name(self) -> str:  # type: ignore[override]
38        """Get name of secret manager."""
39        if self.dotenv_path:
40            return f"{SecretSourceEnum.DOTENV.value}:{self.dotenv_path}"
41        return SecretSourceEnum.DOTENV.value
42
43    def __init__(
44        self,
45        dotenv_path: Path | None = None,
46    ) -> None:
47        """Initialize a new .env Secret Manager, with optionally specified file path."""
48        self.dotenv_path = dotenv_path
49
50    def get_secret(self, secret_name: str) -> SecretString | None:
51        """Get a named secret from the `.env` file."""
52        try:
53            dotenv_vars: dict[str, str | None] = dotenv_values(
54                dotenv_path=self.dotenv_path,
55            )
56        except Exception:
57            # Can't locate or parse a .env file
58            return None
59
60        if secret_name not in dotenv_vars:
61            # Secret not found
62            return None
63
64        return SecretString(dotenv_vars[secret_name])
65
66    def list_secrets_names(self) -> list[str]:
67        """List all secrets available in the .env file."""
68        dotenv_vars: dict[str, str | None] = dotenv_values(
69            dotenv_path=self.dotenv_path,
70        )
71        return list(dotenv_vars.keys())