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

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

40def register_secret_manager(
41    secret_manager: SecretManager,
42    *,
43    as_backup: bool = False,
44    replace_existing: bool = False,
45) -> None:
46    """Register a custom secret manager."""
47    if replace_existing:
48        clear_secret_sources()
49
50    if as_backup:
51        # Add to end of list
52        _SECRETS_SOURCES.append(secret_manager)
53    else:
54        # Add to beginning of list
55        _SECRETS_SOURCES.insert(0, secret_manager)

def clear_secret_sources() -> None: View Source

58def clear_secret_sources() -> None:
59    """Clear all secret sources."""
60    _SECRETS_SOURCES.clear()

Clear all secret sources.

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

63def disable_secret_source(source: SecretManager | SecretSourceEnum) -> None:
64    """Disable one of the default secrets sources.
65
66    This function can accept either a `SecretManager` instance, a `SecretSourceEnum` enum value, or
67    a string representing the name of the source to disable.
68    """
69    if isinstance(source, SecretManager) and source in _SECRETS_SOURCES:
70        _SECRETS_SOURCES.remove(source)
71        return
72
73    # Else, remove by name
74    for existing_source in list(_SECRETS_SOURCES).copy():
75        if str(existing_source) == str(source):
76            _SECRETS_SOURCES.remove(existing_source)

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.