`vendor_fabric.vault`¶

Vault connector built on extended-data primitives.

Submodules¶

vendor_fabric.vault.tools

Package Contents¶

Classes¶

VaultConnector

Vault connector with token and AppRole authentication.

Data¶

`VAULT_URL_ENV_VAR`
`VAULT_NAMESPACE_ENV_VAR`
`VAULT_ROLE_ID_ENV_VAR`
`VAULT_SECRET_ID_ENV_VAR`
`VAULT_APPROLE_PATH_ENV_VAR`

API¶

vendor_fabric.vault.VAULT_URL_ENV_VAR = 'VAULT_ADDR'¶

vendor_fabric.vault.VAULT_NAMESPACE_ENV_VAR = 'VAULT_NAMESPACE'¶

vendor_fabric.vault.VAULT_ROLE_ID_ENV_VAR = 'VAULT_ROLE_ID'¶

vendor_fabric.vault.VAULT_SECRET_ID_ENV_VAR = 'VAULT_SECRET_ID'¶

vendor_fabric.vault.VAULT_APPROLE_PATH_ENV_VAR = 'VAULT_APPROLE_PATH'¶

class vendor_fabric.vault.VaultConnector(vault_url: str | None = None, vault_namespace: str | None = None, vault_token: str | None = None, logger: extended_data.logging.Logging | None = None, **kwargs: Any)¶

Bases: vendor_fabric.base.ConnectorBase

Vault connector with token and AppRole authentication.

Initialization

Initialize the connector.

Args: api_key: API key (overrides environment variable) base_url: Base URL (overrides class default) timeout: HTTP timeout in seconds logger: Logger instance **kwargs: Passed to InputProvider

property vault_client: hvac.Client¶: Lazy initialization of the Vault client.

classmethod get_vault_client(vault_url: str | None = None, vault_namespace: str | None = None, vault_token: str | None = None, **kwargs: Any) → hvac.Client¶: Get an instance of the Vault client.

list_secrets(root_path: str = '/', mount_point: str = 'secret', max_depth: int | None = None) → extended_data.containers.ExtendedDict¶

List secrets recursively from Vault KV v2 engine.

Args: root_path: Starting path for listing (default: “/”). mount_point: KV engine mount point (default: “secret”). max_depth: Maximum directory depth to traverse (None = unlimited).

Returns: Dict mapping secret paths to their data.

Raises: ValueError: If root_path contains path traversal sequences.

read_secret(path: str, mount_point: str = 'secret') → extended_data.containers.ExtendedDict | None¶

Read a single secret from Vault.

Args: path: Path to the secret. mount_point: KV engine mount point (default: “secret”).

Returns: Secret data dict, or None if not found.

get_secret(path: str = '/', secret_name: str | None = None, matchers: dict[str, str] | None = None, mount_point: str = 'secret') → extended_data.containers.ExtendedDict | None¶

Get Vault secret by path, name, or by searching with matchers.

This method supports three modes:

Direct path + secret_name: Fetches secret at path/secret_name
Path with matchers: Searches secrets under path and returns first match
Path without matchers: Returns first non-empty secret found

Args: path: Root path to search or base path for secret_name (default: “/”). secret_name: Specific secret name to append to path. matchers: Dict of key/value pairs to match against secret data. mount_point: KV engine mount point (default: “secret”).

Returns: Secret data dict, or None if not found.

write_secret(path: str, data: dict[str, Any], mount_point: str = 'secret') → bool¶

Write a secret to Vault.

Args: path: Path to write the secret. data: Secret data dict. mount_point: KV engine mount point (default: “secret”).

Returns: True if successful, False otherwise.

list_aws_iam_roles(mount_point: str = 'aws', prefix: str | None = None) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedString]¶

List AWS IAM roles configured in Vault’s AWS secrets engine.

Args: mount_point: AWS secrets engine mount point (default: “aws”). prefix: Optional prefix filter for role names.

Returns: List of role names available for credential generation.

get_aws_iam_role(role_name: str, mount_point: str = 'aws') → extended_data.containers.ExtendedDict | None¶

Retrieve details about a specific AWS IAM role configured in Vault.

Args: role_name: Name of the role to fetch. mount_point: AWS secrets engine mount point (default: “aws”).

Returns: Dict containing the role configuration, or None if not found.

generate_aws_credentials(role_name: str, mount_point: str = 'aws', ttl: str | None = None, credential_type: str | None = None) → extended_data.containers.ExtendedDict¶

Generate AWS credentials via Vault’s AWS secrets engine.

Args: role_name: AWS role configured in Vault. mount_point: AWS secrets engine mount point (default: “aws”). ttl: Optional TTL override (e.g., “1h”). credential_type: Optional credential type override (e.g., “sts”).

Returns: Dict of generated credential data (e.g., AccessKeyId, SecretAccessKey, SessionToken).

Raises: ValueError: If role_name is empty or mount_point is invalid. RuntimeError: If Vault fails to return credentials.

BASE_URL: ClassVar[str] = <Multiline-String>¶

API_KEY_ENV: ClassVar[str] = <Multiline-String>¶

CONNECTOR_CATEGORY: ClassVar[str] = 'external'¶

CONNECTOR_CAPABILITIES: ClassVar[tuple[str, ...]] = ()¶

TIMEOUT: ClassVar[float] = 300.0¶

MIN_REQUEST_INTERVAL: ClassVar[float] = 0.0¶

MAX_RETRIES: ClassVar[int] = 5¶

property api_key: str¶

property client: httpx.Client¶

close() → None¶

request(method: str, endpoint: str, *, headers: dict[str, str] | None = None, **kwargs: Any) → httpx.Response¶

decode_response(response: httpx.Response, *, suffix: str | None = None, as_extended: bool = True) → Any¶

decode_response_file(response: httpx.Response, *, source: str | None = None, suffix: str | None = None, as_extended: bool = True, metadata: collections.abc.Mapping[str, Any] | None = None) → extended_data.io.DataFile¶

extend_result(value: Any) → Any¶

request_data(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

request_data_file(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.io.DataFile¶

request_workflow(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

get(endpoint: str, **kwargs: Any) → httpx.Response¶

get_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

get_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

post(endpoint: str, **kwargs: Any) → httpx.Response¶

post_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

post_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

put(endpoint: str, **kwargs: Any) → httpx.Response¶

put_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

put_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

delete(endpoint: str, **kwargs: Any) → httpx.Response¶

delete_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

delete_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

patch(endpoint: str, **kwargs: Any) → httpx.Response¶

patch_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any¶

patch_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow¶

download(url: str, output_path: str) → int¶

vendor_capabilities: ClassVar[dict[str, vendor_fabric.capabilities.CapabilitySpec]] = None¶

vendor_capability_methods: ClassVar[dict[str, str]] = None¶

get_input(k: str, default: Any | None = None, required: bool = False, is_bool: bool = False, is_integer: bool = False, is_float: bool = False, is_path: bool = False, is_datetime: bool = False, as_extended: bool = False) → Any¶

decode_input(k: str, default: Any | None = None, required: bool = False, decode_from_json: bool = False, decode_from_yaml: bool = False, decode_from_base64: bool = False, allow_none: bool = True, as_extended: bool = False) → Any¶

freeze_inputs() → extended_data.containers.mappings.ExtendedDict¶

thaw_inputs() → extended_data.containers.mappings.ExtendedDict¶

snapshot_inputs(*, frozen: bool = False) → extended_data.containers.mappings.ExtendedDict¶

replace_inputs(new_inputs: collections.abc.Mapping[str, Any] | None, *, clear_frozen: bool = True) → extended_data.containers.mappings.ExtendedDict¶

merge_inputs(new_inputs: collections.abc.Mapping[str, Any] | None) → extended_data.containers.mappings.ExtendedDict¶

shift_inputs() → extended_data.containers.mappings.ExtendedDict¶

vendor_fabric.vault¶