Core¶

Core AWS utilities and common functions used across the library.

AWSServiceAndResourceProvider `dataclass` ¶

AWSServiceAndResourceProvider(service_name)

Bases: AWSServiceProvider[ClientType], Generic[ClientType, ResourceType]

Provider for creating typed AWS service clients and resources.

Extends AWSServiceProvider to also support resource-based access patterns.

Attributes:

Name	Type	Description
`service_name`	`Resources`	The name of the AWS service (must support resources).

get_resource ¶

get_resource(region=None, **kwargs)

Get a typed resource for this AWS service.

Parameters:

Name	Type	Description	Default
`region`	`Optional[str]`	AWS region. Defaults to None (uses default region).	`None`
`**kwargs`		Additional arguments passed to boto3 resource creation.	`{}`

Returns:

Type	Description
`ResourceType`	A typed boto3 resource for the service.

Source code in src/aibs_informatics_aws_utils/core.py

def get_resource(self, region: Optional[str] = None, **kwargs) -> ResourceType:
    """Get a typed resource for this AWS service.

    Args:
        region (Optional[str]): AWS region. Defaults to None (uses default region).
        **kwargs: Additional arguments passed to boto3 resource creation.

    Returns:
        A typed boto3 resource for the service.
    """
    return cast(ResourceType, get_resource(self.service_name, region=region, **kwargs))

AWSServiceProvider `dataclass` ¶

AWSServiceProvider(service_name)

Bases: Generic[ClientType]

Provider for creating typed AWS service clients.

Attributes:

Name	Type	Description
`service_name`	`Services`	The name of the AWS service.

get_client ¶

get_client(region=None, **kwargs)

Get a typed client for this AWS service.

Parameters:

Name	Type	Description	Default
`region`	`Optional[str]`	AWS region. Defaults to None (uses default region).	`None`
`**kwargs`		Additional arguments passed to boto3 client creation.	`{}`

Returns:

Type	Description
`ClientType`	A typed boto3 client for the service.

Source code in src/aibs_informatics_aws_utils/core.py

def get_client(self, region: Optional[str] = None, **kwargs) -> ClientType:
    """Get a typed client for this AWS service.

    Args:
        region (Optional[str]): AWS region. Defaults to None (uses default region).
        **kwargs: Additional arguments passed to boto3 client creation.

    Returns:
        A typed boto3 client for the service.
    """
    return cast(ClientType, get_client(self.service_name, region=region, **kwargs))

client_error_code_check ¶

client_error_code_check(client_error, *error_codes)

Check if a ClientError matches any of the specified error codes.

Parameters:

Name	Type	Description	Default
`client_error`	`ClientError`	The ClientError exception to check.	required
`*error_codes`	`str`	One or more error codes to check against.	`()`

Returns:

Type	Description
`bool`	True if the error code matches any of the specified codes.

Source code in src/aibs_informatics_aws_utils/core.py

def client_error_code_check(client_error: ClientError, *error_codes: str) -> bool:
    """Check if a ClientError matches any of the specified error codes.

    Args:
        client_error (ClientError): The ClientError exception to check.
        *error_codes (str): One or more error codes to check against.

    Returns:
        True if the error code matches any of the specified codes.
    """
    return get_client_error_code(client_error) in error_codes

get_account_id ¶

get_account_id()

Get the AWS account ID from the current credentials.

Returns:

Type	Description
`str`	The AWS account ID as a string.

Source code in src/aibs_informatics_aws_utils/core.py

def get_account_id() -> str:
    """Get the AWS account ID from the current credentials.

    Returns:
        The AWS account ID as a string.
    """
    return get_caller_identity()["Account"]

get_caller_identity ¶

get_caller_identity()

Get the caller identity from AWS STS.

Returns:

Type	Description
`GetCallerIdentityResponseTypeDef`	The caller identity response containing Account, Arn, and UserId.

Source code in src/aibs_informatics_aws_utils/core.py

def get_caller_identity() -> GetCallerIdentityResponseTypeDef:
    """Get the caller identity from AWS STS.

    Returns:
        The caller identity response containing Account, Arn, and UserId.
    """
    return boto3.client("sts").get_caller_identity()

get_client ¶

get_client(service, session=None, region=None, **kwargs)

Get a boto3 client object Notes:

If session is not provided, then the default session is used
Order of priority for region is
region parameter
"region_name" in **kwargs
fallback to whatever is fetched in get_region

Parameters:

Name	Type	Description	Default
`service`	`Clients`	The intended service to build client for	required
`session`	`Session`	A customized session object. Defaults to None.	`None`
`region`	`Optional[str]`	An explicit region. Defaults to None.	`None`

Returns:

Type	Description
	botocore.client.BaseClient: A boto3 Client object

Source code in src/aibs_informatics_aws_utils/core.py

@cache
def get_client(
    service: Clients,
    session: Optional[boto3.Session] = None,
    region: Optional[str] = None,
    **kwargs,
):
    """Get a boto3 client object
    Notes:

    - If session is not provided, then the default session is used
    - Order of priority for region is
      1. region parameter
      2. "region_name" in **kwargs
      3. fallback to whatever is fetched in get_region

    Args:
        service (Clients): The intended service to build client for
        session (boto3.Session, optional): A customized session object. Defaults to None.
        region (Optional[str], optional): An explicit region. Defaults to None.

    Returns:
        botocore.client.BaseClient: A boto3 Client object
    """
    region_name = get_region(region=region or kwargs.get("region_name"))
    if region_name:
        kwargs["region_name"] = region_name

    # If config for our client is not set, we want to set it to use "standard" mode
    # (default is "legacy") and increase the number of retries to 5 (default is 3)
    # See: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/retries.html#available-retry-modes
    config: Optional[Config] = kwargs.pop("config", None)
    default_config = Config(
        connect_timeout=120, read_timeout=120, retries={"max_attempts": 6, "mode": "standard"}
    )
    if config is None:
        config = default_config
    else:
        # Have values in pre-existing config (if it exists) take precedence over default_config
        config = default_config.merge(other_config=config)

    session = session or boto3.Session()
    return session.client(service, config=config, **kwargs)

get_client_error_code ¶

get_client_error_code(client_error)

Extract the error code from a boto3 ClientError.

Parameters:

Name	Type	Description	Default
`client_error`	`ClientError`	The ClientError exception.	required

Returns:

Type	Description
`str`	The error code string, or "Unknown" if not found.

Source code in src/aibs_informatics_aws_utils/core.py

def get_client_error_code(client_error: ClientError) -> str:
    """Extract the error code from a boto3 ClientError.

    Args:
        client_error (ClientError): The ClientError exception.

    Returns:
        The error code string, or "Unknown" if not found.
    """
    return client_error.response.get("Error", {}).get("Code", "Unknown")

get_client_error_message ¶

get_client_error_message(client_error)

Extract the error message from a boto3 ClientError.

Parameters:

Name	Type	Description	Default
`client_error`	`ClientError`	The ClientError exception.	required

Returns:

Type	Description
`str`	The error message string, or "Unknown" if not found.

Source code in src/aibs_informatics_aws_utils/core.py

def get_client_error_message(client_error: ClientError) -> str:
    """Extract the error message from a boto3 ClientError.

    Args:
        client_error (ClientError): The ClientError exception.

    Returns:
        The error message string, or "Unknown" if not found.
    """
    return client_error.response.get("Error", {}).get("Message", "Unknown")

get_iam_arn ¶

get_iam_arn()

Get the IAM ARN from the current credentials.

Returns:

Type	Description
`IAMArn`	The IAM ARN of the current identity.

Source code in src/aibs_informatics_aws_utils/core.py

def get_iam_arn() -> IAMArn:
    """Get the IAM ARN from the current credentials.

    Returns:
        The IAM ARN of the current identity.
    """
    return IAMArn(get_caller_identity()["Arn"])

get_region ¶

get_region(region=None)

Get and sanitize region value

Will retrieve the current region from a newly created boto3 session. This is helpful for getting the region dynamically within a Lambda function or from the environment if that is not available.

Logic of priority for resolving region

User input
boto3.Session
ENV VAR "AWS_REGION"
ENV VAR "REGION"

Subsequently, the region value is validated against a regex

Parameters:

Name	Type	Description	Default
`region`	`str`	user provided region. Defaults to None.	`None`

Raises:

Type	Description
`ApplicationException`	If region cannot be resolved If region is not correct format

Returns:

Type	Description
`str`	AWS Region

Source code in src/aibs_informatics_aws_utils/core.py

def get_region(region: Optional[str] = None) -> str:
    """Get and sanitize region value

    Will retrieve the current region from a newly created boto3 session.
    This is helpful for getting the region dynamically within a Lambda
    function or from the environment if that is not available.

    Logic of priority for resolving region:
        * User input
        * boto3.Session
        * ENV VAR "AWS_REGION"
        * ENV VAR "REGION"

    Subsequently, the region value is validated against a regex

    Args:
        region (str, optional): user provided region. Defaults to None.

    Raises:
        ApplicationException:
            - If region cannot be resolved
            - If region is not correct format

    Returns:
        AWS Region
    """

    # If not provided, check session
    if not region:
        session = Session()
        region = session.region_name
    # If session does not resolve, check environment variables
    if not region:
        # Attempt to get it from an environment variable.
        ENV_VARS = ["AWS_REGION", "REGION"]
        logger.debug(
            "Could not resolve region from session. "
            f"Looking for following env variables: {ENV_VARS}"
        )
        region = next((os.environ.get(key) for key in ENV_VARS if os.environ.get(key)), None)

    if not region:
        error_msg = "Could not determine region from default session or environment"
        logger.error(error_msg)
        raise AWSError(error_msg)
    try:
        region = AWSRegion(region)
        assert region is not None  # mollify mypy
    except Exception as e:
        raise AWSError from e
    return region

get_resource ¶

get_resource(service, session=None, region=None, **kwargs)

Get a boto3 resource object Notes:

If session is not provided, then the default session is used
Order of priority for region is
region parameter
"region_name" in **kwargs
fallback to whatever is fetched in get_region

Parameters:

Name	Type	Description	Default
`service`	`Resources`	The intended service to build resource from	required
`session`	`Session`	A customized session object. Defaults to None.	`None`
`region`	`Optional[str]`	An explicit region. Defaults to None.	`None`

Returns:

Type	Description
`ServiceResource`	boto3.resources.base.ServiceResource: A ServiceResource object

Source code in src/aibs_informatics_aws_utils/core.py

@cache
def get_resource(
    service: Resources,
    session: Optional[boto3.Session] = None,
    region: Optional[str] = None,
    **kwargs,
) -> ServiceResource:
    """Get a boto3 resource object
    Notes:

    - If session is not provided, then the default session is used
    - Order of priority for region is
      1. region parameter
      2. "region_name" in **kwargs
      3. fallback to whatever is fetched in get_region

    Args:
        service (Resources): The intended service to build resource from
        session (boto3.Session, optional): A customized session object. Defaults to None.
        region (Optional[str], optional): An explicit region. Defaults to None.

    Returns:
        boto3.resources.base.ServiceResource: A ServiceResource object
    """
    region_name = get_region(region=region or kwargs.get("region_name"))
    if region_name:
        kwargs["region_name"] = region_name

    # If config for our client is not set, we want to set it to use "standard" mode
    # (default is "legacy") and increase the number of retries to 5 (default is 3)
    # See: https://boto3.amazonaws.com/v1/documentation/api/latest/guide/retries.html#available-retry-modes
    config: Optional[Config] = kwargs.pop("config", None)
    default_config = Config(
        connect_timeout=120, read_timeout=120, retries={"max_attempts": 6, "mode": "standard"}
    )
    if config is None:
        config = default_config
    else:
        # Have values in pre-existing config (if it exists) take precedence over default_config
        config = default_config.merge(other_config=config)

    session = session or boto3.Session()
    return session.resource(service, config=config, **kwargs)

get_session ¶

get_session(session=None)

Get or create a boto3 Session.

Parameters:

Name	Type	Description	Default
`session`	`Optional[Union[Session, Session]]`	An existing Session or BotocoreSession to use. If None, creates a new default Session.	`None`

Returns:

Type	Description
`Session`	A boto3 Session object.

Source code in src/aibs_informatics_aws_utils/core.py

def get_session(session: Optional[Union[Session, BotocoreSession]] = None) -> Session:
    """Get or create a boto3 Session.

    Args:
        session (Optional[Union[Session, BotocoreSession]]): An existing Session
            or BotocoreSession to use. If None, creates a new default Session.

    Returns:
        A boto3 Session object.
    """
    if not session:
        return Session()
    elif isinstance(session, BotocoreSession):
        return Session(botocore_session=session)
    else:
        return session

get_user_id ¶

get_user_id()

Get the IAM user ID from the current credentials.

Returns:

Type	Description
`UserId`	The IAM user ID.

Source code in src/aibs_informatics_aws_utils/core.py

def get_user_id() -> UserId:
    """Get the IAM user ID from the current credentials.

    Returns:
        The IAM user ID.
    """
    return UserId(get_caller_identity()["UserId"])

Core¶

AWSServiceAndResourceProvider dataclass ¶

get_resource ¶

AWSServiceProvider dataclass ¶

get_client ¶

client_error_code_check ¶

get_account_id ¶

get_caller_identity ¶

get_client ¶

get_client_error_code ¶

get_client_error_message ¶

get_iam_arn ¶

get_region ¶

get_resource ¶

get_session ¶

get_user_id ¶

AWSServiceAndResourceProvider `dataclass` ¶

AWSServiceProvider `dataclass` ¶