Attributes

brickflow_plugins.secrets.BRICKFLOW_SECRETS_BACKEND = 'brickflow_secrets_backend' module-attribute

brickflow_plugins.secrets.brickflow_secrets_backend_plugin_impl = pluggy.HookimplMarker(BRICKFLOW_SECRETS_BACKEND) module-attribute

brickflow_plugins.secrets.brickflow_secrets_plugin_spec = pluggy.HookspecMarker(BRICKFLOW_SECRETS_BACKEND) module-attribute

Classes

brickflow_plugins.secrets.AbstractSecretsHelper

Bases: abc.ABC

Attributes

PROTOCOL_STARTS_WITH: Optional[Union[str, List[str]]] = None class-attribute instance-attribute

Functions

get_secret_value_from_url(url_parsed_result: ParseResult)

Source code in brickflow_plugins/secrets/__init__.py

def get_secret_value_from_url(self, url_parsed_result: ParseResult):
    allowed_protocols = (
        [self.PROTOCOL_STARTS_WITH]
        if isinstance(self.PROTOCOL_STARTS_WITH, str)
        else self.PROTOCOL_STARTS_WITH
    )
    if self.PROTOCOL_STARTS_WITH is not None and not any(
        [
            url_parsed_result.scheme.lower().startswith(protocol)
            for protocol in allowed_protocols
        ]
    ):
        return None
    return self._get_secret_value_from_url(url_parsed_result)

brickflow_plugins.secrets.B64SecretsHelper

Bases: AbstractSecretsHelper

Attributes

PROTOCOL_STARTS_WITH = ['base64', 'b64'] class-attribute instance-attribute

brickflow_plugins.secrets.Base64BrickflowSecretPluginImpl

Bases: BrickflowSecretPluginSpec

Functions

get_secret_value(url_parsed_result: ParseResult) -> Optional['str'] staticmethod

Source code in brickflow_plugins/secrets/__init__.py

@staticmethod
@brickflow_secrets_backend_plugin_impl
def get_secret_value(url_parsed_result: ParseResult) -> Optional["str"]:
    return B64SecretsHelper().get_secret_value_from_url(url_parsed_result)

brickflow_plugins.secrets.BrickflowSecretPluginSpec

Functions

get_secret_value(url_parsed_result: ParseResult) -> Optional['str'] staticmethod

Custom execute method that is able to be plugged in.

Source code in brickflow_plugins/secrets/__init__.py

@staticmethod
@brickflow_secrets_plugin_spec(firstresult=True)
def get_secret_value(url_parsed_result: ParseResult) -> Optional["str"]:
    """Custom execute method that is able to be plugged in."""

brickflow_plugins.secrets.BrickflowSecretsBackend

Bases: BaseSecretsBackend

Functions

get_conn_value(conn_id: str) -> str | None

Source code in brickflow_plugins/secrets/__init__.py

def get_conn_value(self, conn_id: str) -> str | None:
    parsed_url = urlparse(conn_id)
    return get_brickflow_tasks_hook().get_secret_value(url_parsed_result=parsed_url)

set_backend_env()

Source code in brickflow_plugins/secrets/__init__.py

def set_backend_env(self):
    for k, v in self._get_secrets_backend_env().items():
        os.environ[k] = v

unset_backend_env()

Source code in brickflow_plugins/secrets/__init__.py

def unset_backend_env(self):
    for k in self._get_secrets_backend_env().keys():
        os.environ.pop(k, None)

brickflow_plugins.secrets.CerberusBrickflowSecretPluginImpl

Bases: BrickflowSecretPluginSpec

Functions

get_secret_value(url_parsed_result: ParseResult) -> Optional['str'] staticmethod

Source code in brickflow_plugins/secrets/__init__.py

@staticmethod
@brickflow_secrets_backend_plugin_impl
def get_secret_value(url_parsed_result: ParseResult) -> Optional["str"]:
    return CerberusSecretsHelper().get_secret_value_from_url(url_parsed_result)

brickflow_plugins.secrets.CerberusSecretsHelper

Bases: AbstractSecretsHelper

Attributes

PROTOCOL_STARTS_WITH = 'cerberus' class-attribute instance-attribute

Functions

parse_path_and_key(path: Optional[str]) -> Optional[Tuple[str, str]] staticmethod

Source code in brickflow_plugins/secrets/__init__.py

@staticmethod
def parse_path_and_key(path: Optional[str]) -> Optional[Tuple[str, str]]:
    if path is not None:
        _cleaned_path = path.lstrip("/").rstrip("/")
        return "/".join(_cleaned_path.split("/")[:-1]), _cleaned_path.split("/")[-1]
    return None

brickflow_plugins.secrets.DatabricksSecretsBrickflowSecretPluginImpl

Bases: BrickflowSecretPluginSpec

Functions

get_secret_value(url_parsed_result: ParseResult) -> Optional['str'] staticmethod

Source code in brickflow_plugins/secrets/__init__.py

@staticmethod
@brickflow_secrets_backend_plugin_impl
def get_secret_value(url_parsed_result: ParseResult) -> Optional["str"]:
    # not implemented yet
    return None

Functions

brickflow_plugins.secrets.get_brickflow_tasks_hook() -> BrickflowSecretPluginSpec cached

Source code in brickflow_plugins/secrets/__init__.py

@functools.lru_cache
def get_brickflow_tasks_hook() -> BrickflowSecretPluginSpec:
    pm = pluggy.PluginManager(BRICKFLOW_SECRETS_BACKEND)
    pm.add_hookspecs(BrickflowSecretPluginSpec)
    pm.load_setuptools_entrypoints(BRICKFLOW_SECRETS_BACKEND)
    pm.register(CerberusBrickflowSecretPluginImpl())
    pm.register(Base64BrickflowSecretPluginImpl())
    for name, plugin_instance in pm.list_name_plugin():
        log.info(
            "Loaded plugin with name: %s and class: %s",
            name,
            plugin_instance.__class__.__name__,
        )
    return pm.hook