vendor_connectors.aws

vendor_connectors.aws

AWS Connector using jbcom ecosystem packages.

This package provides AWS operations organized into submodules:

• organizations: AWS Organizations and Control Tower account management
• sso: IAM Identity Center (SSO) operations
• s3: S3 bucket and object operations
• secrets: Secrets Manager operations (in base connector)
• ecs: ECS cluster and service operations

Usage: from vendor_connectors.aws import AWSConnector

connector = AWSConnector()
accounts = connector.get_accounts()

Submodules

• vendor_connectors.aws.sso
• vendor_connectors.aws.s3
• vendor_connectors.aws.organizations
• vendor_connectors.aws.tools
• vendor_connectors.aws.codedeploy

Package Contents

Classes

AWSConnector	AWS connector for boto3 client and resource management.
AWSConnectorFull	Full AWS connector with all operations.

API

class vendor_connectors.aws.AWSConnector(execution_role_arn: str | None = None, logger: lifecyclelogging.Logging | None = None, **kwargs)

Bases: vendor_connectors.base.VendorConnectorBase

AWS connector for boto3 client and resource management.

This is the base connector class providing:

• Session management and role assumption
• Client/resource creation with retry configuration
• Secrets Manager operations

Higher-level operations are provided via mixin classes from submodules.

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 DirectedInputsClass

assume_role(execution_role_arn: str, role_session_name: str) → boto3.Session

Assume an AWS IAM role and return a boto3 Session.

Args: execution_role_arn: ARN of the role to assume. role_session_name: Name for the assumed role session.

Returns: A boto3 Session with the assumed role credentials.

Raises: RuntimeError: If role assumption fails.

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

Get a boto3 Session, optionally assuming a role.

Args: execution_role_arn: ARN of role to assume. If None, uses default session. role_session_name: Name for the assumed role session.

Returns: A boto3 Session.

static create_standard_retry_config(max_attempts: int = 5) → botocore.config.Config

Create a standard retry configuration.

Args: max_attempts: Maximum retry attempts. Defaults to 5.

Returns: A botocore Config with retry settings.

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) → boto3.client

Get a boto3 client for the specified service.

Args: client_name: AWS service name (e.g., ‘s3’, ‘ec2’, ‘organizations’). execution_role_arn: ARN of role to assume for cross-account access. role_session_name: Name for the assumed role session. config: Optional botocore Config. Defaults to standard retry config. **client_args: Additional arguments passed to boto3 client.

Returns: A boto3 client for the specified service.

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) → boto3.resources.base.ServiceResource

Get a boto3 resource for the specified service.

Args: service_name: AWS service name (e.g., ‘s3’, ‘ec2’, ‘dynamodb’). execution_role_arn: ARN of role to assume for cross-account access. role_session_name: Name for the assumed role session. config: Optional botocore Config. Defaults to standard retry config. **resource_args: Additional arguments passed to boto3 resource.

Returns: A boto3 resource for the specified service.

Raises: RuntimeError: If resource creation fails.

get_caller_account_id() → str

Get the AWS account ID of the caller.

Returns: The 12-digit AWS account ID.

get_secret(secret_id: str, execution_role_arn: str | None = None, role_session_name: str | None = None, secretsmanager: boto3.client | None = None) → str | None

Get a single secret value from AWS Secrets Manager.

Args: secret_id: The ARN or name of the secret to retrieve. execution_role_arn: ARN of role to assume for cross-account access. role_session_name: Session name for assumed role. secretsmanager: Optional pre-existing Secrets Manager client.

Returns: The secret value as a string, or None if not found.

list_secrets(filters: list[dict] | None = None, prefix: str | None = None, get_secret_values: bool = False, skip_empty_secrets: bool = False, execution_role_arn: str | None = None, role_session_name: str | None = None, **kwargs) → dict[str, str | dict]

List secrets from AWS Secrets Manager.

Args: filters: List of filter dicts for list_secrets API. prefix: Optional prefix for the AWS “name” filter. get_secret_values: If True, fetch actual secret values. skip_empty_secrets: If True, skip secrets with empty values. execution_role_arn: ARN of role to assume for cross-account access. role_session_name: Session name for assumed role. **kwargs: Support for ‘name_prefix’ alias.

Returns: Dict mapping secret names to ARNs or values.

Raises: ValueError: If prefix contains invalid characters.

create_secret(name: str, secret_value: str, description: str = ”, tags: dict[str, str] | None = None, execution_role_arn: str | None = None) → dict[str, Any]

Create a new secret in AWS Secrets Manager.

update_secret(secret_id: str, secret_value: str, execution_role_arn: str | None = None) → dict[str, Any]

Update an existing secret value.

delete_secret(secret_id: str, force_delete: bool = False, recovery_window_days: int = 30, execution_role_arn: str | None = None) → dict[str, Any]

Delete a secret from AWS Secrets Manager.

delete_secrets_matching(prefix: str | None = None, force_delete: bool = False, dry_run: bool = True, execution_role_arn: str | None = None, **kwargs) → list[str]

Delete all secrets that match the provided name prefix.

copy_secrets_to_s3(secrets: dict[str, str | dict], bucket: str, key: str, execution_role_arn: str | None = None, role_session_name: str | None = None) → str

Copy secrets dictionary to S3 as JSON.

Args: secrets: Dictionary of secrets to upload. bucket: S3 bucket name. key: S3 object key. execution_role_arn: ARN of role to assume for S3 access. role_session_name: Session name for assumed role.

Returns: S3 URI of uploaded object.

static load_vendors_from_asm(prefix: str = ‘/vendors/’) → dict[str, str]

Load vendor secrets from AWS Secrets Manager.

This is used in Lambda environments where vendor credentials are stored in ASM under a common prefix (e.g., /vendors/).

Args: prefix: The prefix path for vendor secrets (default: /vendors/)

Returns: Dictionary mapping secret keys (with prefix removed) to their values.

class vendor_connectors.aws.AWSConnectorFull(execution_role_arn: str | None = None, logger: lifecyclelogging.Logging | None = None, **kwargs)

Bases: vendor_connectors.aws.AWSConnector, vendor_connectors.aws.organizations.AWSOrganizationsMixin, vendor_connectors.aws.sso.AWSSSOmixin, vendor_connectors.aws.s3.AWSS3Mixin

Full AWS connector with all operations.

This class combines the base AWSConnector with all operation mixins. Use this for full functionality, or use AWSConnector directly and import specific mixins as needed.

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 DirectedInputsClass