vendor_fabric.connectors

ConnectorFabric - cached connector access for vendor fabric.

Module Contents

Classes

ConnectorFabric

Public API for vendor fabric with client caching.

API

class vendor_fabric.connectors.ConnectorFabric(logger: extended_data.logging.Logging | None = None, **kwargs: Any)

Bases: extended_data.inputs.InputProvider

Public API for vendor fabric with client caching.

This class provides cached access to registered connectors while sharing input snapshots, lifecycle logging, and data normalization.

Usage: vc = ConnectorFabric() slack = vc.get_slack_client(token=”…”, bot_token=”…”) github = vc.get_github_client(github_owner=”org”, github_token=”…”) aws_client = vc.get_aws_client(“s3”)

For Meshy AI, use the functional interface directly: from vendor_fabric.meshy import text3d, image3d, rigging, animate model = text3d.generate(“a medieval sword”)

Meshy does not provide a get_meshy_client() method because it uses a functional interface rather than a connector class, in order to simplify async operations and usage.

Initialization

Initializes the InputProvider with the provided inputs.

Optionally loading additional inputs from environment variables and stdin.

Args: inputs (Mapping[str, Any] | None): Initial inputs to be processed. from_environment (bool): Whether to load inputs from environment variables. from_stdin (bool): Whether to load inputs from stdin. env_prefix (str | None): Optional prefix to filter environment variables. strip_env_prefix (bool): Whether to strip the prefix from environment keys.

list_connectors() → extended_data.containers.ExtendedList[extended_data.containers.ExtendedString]

List connector catalog names.

list_available_connectors() → extended_data.containers.ExtendedList[extended_data.containers.ExtendedString]

List connector names available in the current environment.

list_connector_info(*, include_unavailable: bool = True) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List connector catalog metadata.

list_connector_categories(*, include_unavailable: bool = True) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedString]

List connector catalog categories.

list_connector_capabilities(*, include_unavailable: bool = True) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedString]

List connector catalog capabilities.

list_connectors_by_category(category: str, *, include_unavailable: bool = True) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List connector catalog metadata for a category.

list_connectors_by_capability(capability: str, *, include_unavailable: bool = True) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

List connector catalog metadata for a capability.

get_connector_info(name: str, *, include_unavailable: bool = True) → extended_data.containers.ExtendedDict

Get catalog metadata for one connector.

get_connector_adapter(name: str, *, include_unavailable: bool = True) → vendor_fabric.registry.ConnectorAdapter

Get the adapter registry entry for one connector.

get_connector(name: str, **kwargs: Any) → vendor_fabric.base.ConnectorBase

Get a cached connector instance by registry name.

The connector receives the fabric’s shared inputs and logger unless explicit values are passed in kwargs. This is the generic path for connectors that are registered through entry points or built-ins.

get_aws_connector(execution_role_arn: str | None = None) → vendor_fabric.aws.AWSConnector

Get a cached AWSConnector instance.

Requires: pip install vendor-fabric[aws]

get_aws_client(client_name: str, execution_role_arn: str | None = None, role_session_name: str | None = None, config: botocore.config.Config | None = None, **client_args: Any) → boto3.client

Get a cached boto3 client.

get_aws_resource(service_name: str, execution_role_arn: str | None = None, role_session_name: str | None = None, config: botocore.config.Config | None = None, **resource_args: Any) → boto3.resources.base.ServiceResource

Get a cached boto3 resource.

get_aws_session(execution_role_arn: str | None = None, role_session_name: str | None = None) → boto3.Session

Get a cached boto3 session.

get_github_client(github_owner: str | None = None, github_repo: str | None = None, github_branch: str | None = None, github_token: str | None = None) → vendor_fabric.github.GitHubConnector

Get a cached GitHubConnector instance.

Requires: pip install vendor-fabric[github]

get_google_client(service_account_info: dict[str, Any] | str | None = None, scopes: list[str] | None = None, subject: str | None = None) → vendor_fabric.google.GoogleConnector

Get a cached GoogleConnector instance.

Requires: pip install vendor-fabric[google]

get_slack_client(token: str | None = None, bot_token: str | None = None) → vendor_fabric.slack.SlackConnector

Get a cached SlackConnector instance.

Requires: pip install vendor-fabric[slack]

get_vault_client(vault_url: str | None = None, vault_namespace: str | None = None, vault_token: str | None = None) → hvac.Client

Get a cached Vault hvac.Client instance.

Requires: pip install vendor-fabric[vault]

get_vault_connector(vault_url: str | None = None, vault_namespace: str | None = None, vault_token: str | None = None) → vendor_fabric.vault.VaultConnector

Get a cached VaultConnector instance.

Requires: pip install vendor-fabric[vault]

get_zoom_client(client_id: str | None = None, client_secret: str | None = None, account_id: str | None = None) → vendor_fabric.zoom.ZoomConnector

Get a cached ZoomConnector instance.

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