airbyte.secrets.config

This module provides customization of how PyAirbyte locates secrets.

View Source

 1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 2"""This module provides customization of how PyAirbyte locates secrets."""
 3
 4from __future__ import annotations
 5
 6from typing import TYPE_CHECKING
 7
 8from airbyte._util import meta
 9from airbyte.secrets.base import SecretManager
10from airbyte.secrets.env_vars import DotenvSecretManager, EnvVarSecretManager
11from airbyte.secrets.google_colab import ColabSecretManager
12from airbyte.secrets.prompt import SecretsPrompt
13
14
15if TYPE_CHECKING:
16    from airbyte.secrets.base import SecretSourceEnum
17    from airbyte.secrets.custom import CustomSecretManager
18
19
20_SECRETS_SOURCES: list[SecretManager] = []
21
22
23def _get_secret_sources() -> list[SecretManager]:
24    """Initialize the default secret sources."""
25    if len(_SECRETS_SOURCES) == 0:
26        # Initialize the default secret sources
27        _SECRETS_SOURCES.extend(
28            [
29                EnvVarSecretManager(),
30                DotenvSecretManager(),
31            ]
32        )
33        if meta.is_colab():
34            _SECRETS_SOURCES.append(ColabSecretManager())
35
36        if meta.is_interactive():
37            _SECRETS_SOURCES.append(SecretsPrompt())
38
39    return _SECRETS_SOURCES.copy()
40
41
42# Ensure the default secret sources are initialized
43_ = _get_secret_sources()
44
45
46def register_secret_manager(
47    secret_manager: CustomSecretManager,
48    *,
49    as_backup: bool = False,
50    replace_existing: bool = False,
51) -> None:
52    """Register a custom secret manager."""
53    if replace_existing:
54        clear_secret_sources()
55
56    if as_backup:
57        # Add to end of list
58        _SECRETS_SOURCES.append(secret_manager)
59    else:
60        # Add to beginning of list
61        _SECRETS_SOURCES.insert(0, secret_manager)
62
63
64def clear_secret_sources() -> None:
65    """Clear all secret sources."""
66    _SECRETS_SOURCES.clear()
67
68
69def disable_secret_source(source: SecretManager | SecretSourceEnum) -> None:
70    """Disable one of the default secrets sources.
71
72    This function can accept either a `SecretManager` instance, a `SecretSourceEnum` enum value, or
73    a string representing the name of the source to disable.
74    """
75    if isinstance(source, SecretManager) and source in _SECRETS_SOURCES:
76        _SECRETS_SOURCES.remove(source)
77        return
78
79    # Else, remove by name
80    for s in list(_SECRETS_SOURCES).copy():
81        if s.name == str(source):
82            _SECRETS_SOURCES.remove(s)

def register_secret_manager( secret_manager: airbyte.secrets.CustomSecretManager, *, as_backup: bool = False, replace_existing: bool = False) -> None: View Source

47def register_secret_manager(
48    secret_manager: CustomSecretManager,
49    *,
50    as_backup: bool = False,
51    replace_existing: bool = False,
52) -> None:
53    """Register a custom secret manager."""
54    if replace_existing:
55        clear_secret_sources()
56
57    if as_backup:
58        # Add to end of list
59        _SECRETS_SOURCES.append(secret_manager)
60    else:
61        # Add to beginning of list
62        _SECRETS_SOURCES.insert(0, secret_manager)

def clear_secret_sources() -> None: View Source

65def clear_secret_sources() -> None:
66    """Clear all secret sources."""
67    _SECRETS_SOURCES.clear()

Clear all secret sources.

def disable_secret_source( source: airbyte.secrets.SecretManager | airbyte.SecretSourceEnum) -> None: View Source

70def disable_secret_source(source: SecretManager | SecretSourceEnum) -> None:
71    """Disable one of the default secrets sources.
72
73    This function can accept either a `SecretManager` instance, a `SecretSourceEnum` enum value, or
74    a string representing the name of the source to disable.
75    """
76    if isinstance(source, SecretManager) and source in _SECRETS_SOURCES:
77        _SECRETS_SOURCES.remove(source)
78        return
79
80    # Else, remove by name
81    for s in list(_SECRETS_SOURCES).copy():
82        if s.name == str(source):
83            _SECRETS_SOURCES.remove(s)

Disable one of the default secrets sources.

This function can accept either a SecretManager instance, a SecretSourceEnum enum value, or a string representing the name of the source to disable.