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

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

46def register_secret_manager(
47    secret_manager: SecretManager,
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)

def clear_secret_sources() -> None: View Source

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

Clear all secret sources.

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

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)

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.