AWS Utilities¶

AWS helper functions and utilities for CDK constructs.

IAM Utilities¶

iam_utils ¶

IAM utilities for building policy statements and granting permissions.

This module provides predefined IAM actions lists and helper functions for creating policy statements for various AWS services.

Note

The list of actions for each service is incomplete and based on project needs. A helpful resource to research actions is: https://www.awsiamactions.io/

Functions¶

grant_managed_policies ¶

grant_managed_policies(
    role: IRole | None,
    *managed_policies: str | ManagedPolicy
) -> None

Grant managed policies to an IAM role.

Parameters:

Name	Type	Description	Default
`role`	`Optional[IRole]`	The IAM role to grant policies to. If None, no action is taken.	required
`*managed_policies`	`Union[str, ManagedPolicy]`	Variable number of managed policies to grant. Can be policy names (str) or ManagedPolicy objects.	`()`

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def grant_managed_policies(
    role: iam.IRole | None,
    *managed_policies: str | iam.ManagedPolicy,
) -> None:
    """Grant managed policies to an IAM role.

    Args:
        role (Optional[iam.IRole]): The IAM role to grant policies to.
            If None, no action is taken.
        *managed_policies (Union[str, iam.ManagedPolicy]): Variable number of
            managed policies to grant. Can be policy names (str) or
            ManagedPolicy objects.
    """
    if not role:
        return

    for mp in managed_policies:
        role.add_managed_policy(
            iam.ManagedPolicy.from_aws_managed_policy_name(mp) if isinstance(mp, str) else mp
        )

batch_policy_statement ¶

batch_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = BATCH_FULL_ACCESS_ACTIONS,
    sid: str = "BatchReadWrite",
) -> PolicyStatement

Create an IAM policy statement for AWS Batch.

Parameters:

Name	Type	Description	Default
`env_base`	`Optional[EnvBase]`	Environment base for resource prefix. Defaults to None (matches all).	`None`
`actions`	`List[str]`	List of Batch actions to allow. Defaults to BATCH_FULL_ACCESS_ACTIONS.	`BATCH_FULL_ACCESS_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "BatchReadWrite".	`'BatchReadWrite'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for Batch resources.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def batch_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = BATCH_FULL_ACCESS_ACTIONS,
    sid: str = "BatchReadWrite",
) -> iam.PolicyStatement:
    """Create an IAM policy statement for AWS Batch.

    Args:
        env_base (Optional[EnvBase]): Environment base for resource prefix.
            Defaults to None (matches all).
        actions (List[str]): List of Batch actions to allow.
            Defaults to BATCH_FULL_ACCESS_ACTIONS.
        sid (str): Statement ID. Defaults to "BatchReadWrite".

    Returns:
        IAM policy statement for Batch resources.
    """
    resource_id = f"{env_base or ''}*"

    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_batch_arn(
                resource_id=resource_id,
                resource_type="compute-environment",
            ),
            build_batch_arn(
                resource_id=resource_id,
                resource_type="job",
            ),
            build_batch_arn(
                resource_id=resource_id,
                resource_type="job-definition",
            ),
            build_batch_arn(
                resource_id=resource_id,
                resource_type="job-queue",
            ),
            # ERROR: An error occurred (AccessDeniedException) when calling the
            # DescribeJobDefinitions operation:
            # User: arn:aws:sts::051791135335:assumed-role/Infrastructure.../dev-ryan-gwo-create-job-definition-fn  # noqa: E501
            # is not authorized to perform: batch:DescribeJobDefinitions on resource: "*"
            # TODO: WTF why does this not work... adding "*" resource for now
            "*",
        ],
    )

dynamodb_policy_statement ¶

dynamodb_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = DYNAMODB_READ_WRITE_ACTIONS,
    sid: str = "DynamoDBReadWrite",
) -> PolicyStatement

Create an IAM policy statement for DynamoDB.

Parameters:

Name	Type	Description	Default
`env_base`	`Optional[EnvBase]`	Environment base for resource prefix. Defaults to None (matches all).	`None`
`actions`	`List[str]`	List of DynamoDB actions to allow. Defaults to DYNAMODB_READ_WRITE_ACTIONS.	`DYNAMODB_READ_WRITE_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "DynamoDBReadWrite".	`'DynamoDBReadWrite'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for DynamoDB tables.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def dynamodb_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = DYNAMODB_READ_WRITE_ACTIONS,
    sid: str = "DynamoDBReadWrite",
) -> iam.PolicyStatement:
    """Create an IAM policy statement for DynamoDB.

    Args:
        env_base (Optional[EnvBase]): Environment base for resource prefix.
            Defaults to None (matches all).
        actions (List[str]): List of DynamoDB actions to allow.
            Defaults to DYNAMODB_READ_WRITE_ACTIONS.
        sid (str): Statement ID. Defaults to "DynamoDBReadWrite".

    Returns:
        IAM policy statement for DynamoDB tables.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_dynamodb_arn(
                resource_id=f"{env_base or ''}*",
                resource_type="table",
            ),
        ],
    )

ecs_policy_statement ¶

ecs_policy_statement(
    actions: list[str] = ECS_READ_ACTIONS,
    sid: str = "ECSDescribe",
) -> PolicyStatement

Create an IAM policy statement for ECS.

Parameters:

Name	Type	Description	Default
`actions`	`List[str]`	List of ECS actions to allow. Defaults to ECS_READ_ACTIONS.	`ECS_READ_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "ECSDescribe".	`'ECSDescribe'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for ECS resources.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def ecs_policy_statement(
    actions: list[str] = ECS_READ_ACTIONS, sid: str = "ECSDescribe"
) -> iam.PolicyStatement:
    """Create an IAM policy statement for ECS.

    Args:
        actions (List[str]): List of ECS actions to allow.
            Defaults to ECS_READ_ACTIONS.
        sid (str): Statement ID. Defaults to "ECSDescribe".

    Returns:
        IAM policy statement for ECS resources.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_arn(
                service="ecs",
                resource_id="*/*",
                resource_type="container-instance",
                resource_delim="/",
            ),
        ],
    )

lambda_policy_statement ¶

lambda_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = LAMBDA_FULL_ACCESS_ACTIONS,
    sid: str = "LambdaReadWrite",
) -> PolicyStatement

Create an IAM policy statement for Lambda.

Parameters:

Name	Type	Description	Default
`env_base`	`Optional[EnvBase]`	Environment base for resource prefix. Defaults to None (matches all).	`None`
`actions`	`List[str]`	List of Lambda actions to allow. Defaults to LAMBDA_FULL_ACCESS_ACTIONS.	`LAMBDA_FULL_ACCESS_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "LambdaReadWrite".	`'LambdaReadWrite'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for Lambda functions.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def lambda_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = LAMBDA_FULL_ACCESS_ACTIONS,
    sid: str = "LambdaReadWrite",
) -> iam.PolicyStatement:
    """Create an IAM policy statement for Lambda.

    Args:
        env_base (Optional[EnvBase]): Environment base for resource prefix.
            Defaults to None (matches all).
        actions (List[str]): List of Lambda actions to allow.
            Defaults to LAMBDA_FULL_ACCESS_ACTIONS.
        sid (str): Statement ID. Defaults to "LambdaReadWrite".

    Returns:
        IAM policy statement for Lambda functions.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_lambda_arn(
                resource_id=f"{env_base or ''}*",
                resource_type="function",
            ),
        ],
    )

s3_policy_statement ¶

s3_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = S3_FULL_ACCESS_ACTIONS,
    sid: str = "S3FullAccess",
) -> PolicyStatement

Create an IAM policy statement for S3.

Parameters:

Name	Type	Description	Default
`env_base`	`Optional[EnvBase]`	Environment base for resource prefix. Defaults to None (matches all).	`None`
`actions`	`List[str]`	List of S3 actions to allow. Defaults to S3_FULL_ACCESS_ACTIONS.	`S3_FULL_ACCESS_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "S3FullAccess".	`'S3FullAccess'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for S3 buckets.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def s3_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = S3_FULL_ACCESS_ACTIONS,
    sid: str = "S3FullAccess",
) -> iam.PolicyStatement:
    """Create an IAM policy statement for S3.

    Args:
        env_base (Optional[EnvBase]): Environment base for resource prefix.
            Defaults to None (matches all).
        actions (List[str]): List of S3 actions to allow.
            Defaults to S3_FULL_ACCESS_ACTIONS.
        sid (str): Statement ID. Defaults to "S3FullAccess".

    Returns:
        IAM policy statement for S3 buckets.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_s3_arn(
                resource_id=f"{env_base or ''}*",
                resource_type="bucket",
            ),
        ],
    )

secretsmanager_policy_statement ¶

secretsmanager_policy_statement(
    actions: list[str] = SECRETSMANAGER_READ_ONLY_ACTIONS,
    sid: str = "SecretsManagerReadOnly",
    resource_id: str = "*",
    region: str = None,
    account: str = None,
) -> PolicyStatement

Create an IAM policy statement for Secrets Manager.

Parameters:

Name	Type	Description	Default
`actions`	`List[str]`	List of Secrets Manager actions to allow. Defaults to SECRETSMANAGER_READ_ONLY_ACTIONS.	`SECRETSMANAGER_READ_ONLY_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "SecretsManagerReadOnly".	`'SecretsManagerReadOnly'`
`resource_id`	`str`	Resource identifier. Defaults to "*".	`'*'`
`region`	`str`	AWS region. Defaults to None (current region).	`None`
`account`	`str`	AWS account ID. Defaults to None (current account).	`None`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for Secrets Manager resources.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def secretsmanager_policy_statement(
    actions: list[str] = SECRETSMANAGER_READ_ONLY_ACTIONS,
    sid: str = "SecretsManagerReadOnly",
    resource_id: str = "*",
    region: str = None,
    account: str = None,
) -> iam.PolicyStatement:
    """Create an IAM policy statement for Secrets Manager.

    Args:
        actions (List[str]): List of Secrets Manager actions to allow.
            Defaults to SECRETSMANAGER_READ_ONLY_ACTIONS.
        sid (str): Statement ID. Defaults to "SecretsManagerReadOnly".
        resource_id (str): Resource identifier. Defaults to "*".
        region (str): AWS region. Defaults to None (current region).
        account (str): AWS account ID. Defaults to None (current account).

    Returns:
        IAM policy statement for Secrets Manager resources.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_arn(
                service="secretsmanager",
                resource_id=resource_id,
                region=region,
                account=account,
            ),
        ],
    )

ses_policy_statement ¶

ses_policy_statement(
    actions: list[str] = SES_FULL_ACCESS_ACTIONS,
    sid: str = "SESFullAccess",
) -> PolicyStatement

Create an IAM policy statement for SES.

Parameters:

Name	Type	Description	Default
`actions`	`List[str]`	List of SES actions to allow. Defaults to SES_FULL_ACCESS_ACTIONS.	`SES_FULL_ACCESS_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "SESFullAccess".	`'SESFullAccess'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for SES resources.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def ses_policy_statement(
    actions: list[str] = SES_FULL_ACCESS_ACTIONS,
    sid: str = "SESFullAccess",
) -> iam.PolicyStatement:
    """Create an IAM policy statement for SES.

    Args:
        actions (List[str]): List of SES actions to allow.
            Defaults to SES_FULL_ACCESS_ACTIONS.
        sid (str): Statement ID. Defaults to "SESFullAccess".

    Returns:
        IAM policy statement for SES resources.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_arn(
                service="ses",
            ),
        ],
    )

sfn_policy_statement ¶

sfn_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = SFN_STATES_READ_ACCESS_ACTIONS,
    sid: str = "SfnFullAccess",
) -> PolicyStatement

Create an IAM policy statement for Step Functions.

Parameters:

Name	Type	Description	Default
`env_base`	`Optional[EnvBase]`	Environment base for resource prefix. Defaults to None (matches all).	`None`
`actions`	`List[str]`	List of Step Functions actions to allow. Defaults to SFN_STATES_READ_ACCESS_ACTIONS.	`SFN_STATES_READ_ACCESS_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "SfnFullAccess".	`'SfnFullAccess'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for Step Functions resources.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def sfn_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = SFN_STATES_READ_ACCESS_ACTIONS,
    sid: str = "SfnFullAccess",
) -> iam.PolicyStatement:
    """Create an IAM policy statement for Step Functions.

    Args:
        env_base (Optional[EnvBase]): Environment base for resource prefix.
            Defaults to None (matches all).
        actions (List[str]): List of Step Functions actions to allow.
            Defaults to SFN_STATES_READ_ACCESS_ACTIONS.
        sid (str): Statement ID. Defaults to "SfnFullAccess".

    Returns:
        IAM policy statement for Step Functions resources.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_sfn_arn(
                resource_id=f"{env_base or ''}*",
                resource_type="*",
            ),
        ],
    )

sns_policy_statement ¶

sns_policy_statement(
    actions: list[str] = SNS_FULL_ACCESS_ACTIONS,
    sid: str = "SNSFullAccess",
) -> PolicyStatement

Create an IAM policy statement for SNS.

Parameters:

Name	Type	Description	Default
`actions`	`List[str]`	List of SNS actions to allow. Defaults to SNS_FULL_ACCESS_ACTIONS.	`SNS_FULL_ACCESS_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "SNSFullAccess".	`'SNSFullAccess'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for SNS resources.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def sns_policy_statement(
    actions: list[str] = SNS_FULL_ACCESS_ACTIONS,
    sid: str = "SNSFullAccess",
) -> iam.PolicyStatement:
    """Create an IAM policy statement for SNS.

    Args:
        actions (List[str]): List of SNS actions to allow.
            Defaults to SNS_FULL_ACCESS_ACTIONS.
        sid (str): Statement ID. Defaults to "SNSFullAccess".

    Returns:
        IAM policy statement for SNS resources.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_arn(
                service="sns",
            ),
        ],
    )

ssm_policy_statement ¶

ssm_policy_statement(
    actions: list[str] = SSM_READ_ACTIONS,
    sid: str = "SSMParamReadActions",
) -> PolicyStatement

Create an IAM policy statement for SSM Parameter Store.

Parameters:

Name	Type	Description	Default
`actions`	`List[str]`	List of SSM actions to allow. Defaults to SSM_READ_ACTIONS.	`SSM_READ_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "SSMParamReadActions".	`'SSMParamReadActions'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for SSM resources.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def ssm_policy_statement(
    actions: list[str] = SSM_READ_ACTIONS, sid: str = "SSMParamReadActions"
) -> iam.PolicyStatement:
    """Create an IAM policy statement for SSM Parameter Store.

    Args:
        actions (List[str]): List of SSM actions to allow.
            Defaults to SSM_READ_ACTIONS.
        sid (str): Statement ID. Defaults to "SSMParamReadActions".

    Returns:
        IAM policy statement for SSM resources.
    """
    return iam.PolicyStatement(
        sid=sid, actions=actions, effect=iam.Effect.ALLOW, resources=[build_arn(service="ssm")]
    )

sqs_policy_statement ¶

sqs_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = SQS_FULL_ACCESS_ACTIONS,
    sid: str = "SQSFullAccess",
) -> PolicyStatement

Create an IAM policy statement for SQS.

Parameters:

Name	Type	Description	Default
`env_base`	`Optional[EnvBase]`	Environment base for resource prefix. Defaults to None (matches all).	`None`
`actions`	`List[str]`	List of SQS actions to allow. Defaults to SQS_FULL_ACCESS_ACTIONS.	`SQS_FULL_ACCESS_ACTIONS`
`sid`	`str`	Statement ID. Defaults to "SQSFullAccess".	`'SQSFullAccess'`

Returns:

Type	Description
`PolicyStatement`	IAM policy statement for SQS queues.

Source code in src/aibs_informatics_cdk_lib/common/aws/iam_utils.py

def sqs_policy_statement(
    env_base: EnvBase | None = None,
    actions: list[str] = SQS_FULL_ACCESS_ACTIONS,
    sid: str = "SQSFullAccess",
) -> iam.PolicyStatement:
    """Create an IAM policy statement for SQS.

    Args:
        env_base (Optional[EnvBase]): Environment base for resource prefix.
            Defaults to None (matches all).
        actions (List[str]): List of SQS actions to allow.
            Defaults to SQS_FULL_ACCESS_ACTIONS.
        sid (str): Statement ID. Defaults to "SQSFullAccess".

    Returns:
        IAM policy statement for SQS queues.
    """
    return iam.PolicyStatement(
        sid=sid,
        actions=actions,
        effect=iam.Effect.ALLOW,
        resources=[
            build_arn(
                service="sqs",
                resource_id=f"{env_base or ''}*",
            )
        ],
    )

Step Functions Utilities¶

sfn_utils ¶

Step Functions utilities for JSON path references.

This module provides utilities for working with Step Functions JSON path expressions.

Classes¶

JsonReferencePath ¶

Bases: str

String extension for defining JsonPath reference expressions.

Provides properties and methods for working with JSON path references in AWS Step Functions state machines.

More details: https://github.com/json-path/JsonPath

Primarily supports "$" reference.

Example

path = JsonReferencePath("input.data") path.as_reference '$.input.data' path.as_key 'input.data.$'

Attributes¶

as_key `property` ¶

as_key: str

Return the reference path as a key.

Appends ".$" suffix for use as a state machine key.

Returns:

Type	Description
`str`	The path formatted as a key.

as_reference `property` ¶

as_reference: str

Return the reference path as a value.

Prepends "$." prefix for use as a state machine reference.

Returns:

Type	Description
`str`	The path formatted as a reference.

as_jsonpath_string `property` ¶

as_jsonpath_string: str

Return the path as a Step Functions string reference.

Returns:

Type	Description
`str`	The path wrapped in JsonPath.string_at().

as_jsonpath_object `property` ¶

as_jsonpath_object: IResolvable

Return the path as a Step Functions object reference.

Returns:

Type	Description
`IResolvable`	The path wrapped in JsonPath.object_at().

as_jsonpath_json_to_string `property` ¶

as_jsonpath_json_to_string: str

Return the path as a JSON-to-string conversion.

Returns:

Type	Description
`str`	The object reference converted to string via JsonPath.json_to_string().

as_jsonpath_list `property` ¶

as_jsonpath_list: list[str]

Return the path as a Step Functions list reference.

Returns:

Type	Description
`list[str]`	The path wrapped in JsonPath.list_at().

as_jsonpath_number `property` ¶

as_jsonpath_number: int | float

Return the path as a Step Functions number reference.

Returns:

Type	Description
`int \| float`	The path wrapped in JsonPath.number_at().

Functions¶

new ¶

__new__(content: str)

Create a new JsonReferencePath.

Parameters:

Name	Type	Description	Default
`content`	`str`	The path content to wrap.	required

Returns:

Type	Description
	A sanitized JsonReferencePath instance.

Source code in src/aibs_informatics_cdk_lib/common/aws/sfn_utils.py

def __new__(cls, content: str):
    """Create a new JsonReferencePath.

    Args:
        content (str): The path content to wrap.

    Returns:
        A sanitized JsonReferencePath instance.
    """
    return super().__new__(cls, cls.sanitize(content))

add ¶

__add__(other)

Concatenate paths with a period separator.

Parameters:

Name	Type	Description	Default
`other`		The path segment to append.	required

Returns:

Type	Description
	A new JsonReferencePath with the appended segment.

Source code in src/aibs_informatics_cdk_lib/common/aws/sfn_utils.py

def __add__(self, other):
    """Concatenate paths with a period separator.

    Args:
        other: The path segment to append.

    Returns:
        A new JsonReferencePath with the appended segment.
    """
    return JsonReferencePath(super().__add__("." + other))

extend ¶

extend(*paths: str) -> JsonReferencePath

Extend the path with multiple segments.

Parameters:

Name	Type	Description	Default
`*paths`	`str`	Variable number of path segments to append.	`()`

Returns:

Type	Description
`JsonReferencePath`	A new JsonReferencePath with all segments appended.

Source code in src/aibs_informatics_cdk_lib/common/aws/sfn_utils.py

def extend(self, *paths: str) -> "JsonReferencePath":
    """Extend the path with multiple segments.

    Args:
        *paths (str): Variable number of path segments to append.

    Returns:
        A new JsonReferencePath with all segments appended.
    """
    return cast(JsonReferencePath, reduce(lambda x, y: x + y, [self, *paths]))

sanitize `classmethod` ¶

sanitize(s: str) -> str

Sanitize a string for use as a JSON path.

Ensures string has non-consecutive periods and no periods at edges.

Parameters:

Name	Type	Description	Default
`s`	`str`	The string to sanitize.	required

Returns:

Type	Description
`str`	The sanitized string.

Source code in src/aibs_informatics_cdk_lib/common/aws/sfn_utils.py

@classmethod
def sanitize(cls, s: str) -> str:
    """Sanitize a string for use as a JSON path.

    Ensures string has non-consecutive periods and no periods at edges.

    Args:
        s (str): The string to sanitize.

    Returns:
        The sanitized string.
    """
    return f"{cls._EXTRA_PERIODS_PATTERN.sub('.', s).strip('.')}"

is_reference `classmethod` ¶

is_reference(s: Any) -> bool

Check if a value is a JSON path reference.

Parameters:

Name	Type	Description	Default
`s`	`Any`	The value to check.	required

Returns:

Type	Description
`bool`	True if the value is a JsonReferencePath or starts with "$".

Source code in src/aibs_informatics_cdk_lib/common/aws/sfn_utils.py

@classmethod
def is_reference(cls, s: Any) -> bool:
    """Check if a value is a JSON path reference.

    Args:
        s (Any): The value to check.

    Returns:
        True if the value is a JsonReferencePath or starts with "$".
    """
    return isinstance(s, JsonReferencePath) or isinstance(s, str) and s.startswith("$")

empty `classmethod` ¶

empty() -> JsonReferencePath

Create an empty JsonReferencePath.

Returns:

Type	Description
`JsonReferencePath`	An empty JsonReferencePath instance.

Source code in src/aibs_informatics_cdk_lib/common/aws/sfn_utils.py

@classmethod
def empty(cls) -> "JsonReferencePath":
    """Create an empty JsonReferencePath.

    Returns:
        An empty JsonReferencePath instance.
    """
    return cls("")

Core Utilities¶

core_utils ¶

Core AWS utility functions for building ARNs.

This module provides functions for constructing AWS ARNs for various services.

Functions¶

build_arn ¶

build_arn(
    partition: str = "aws",
    service: str | None = None,
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: str | None = None,
    resource_delim: Literal["/", ":"] = ":",
) -> str

Build an AWS ARN string.

Parameters:

Name	Type	Description	Default
`partition`	`str`	AWS partition. Defaults to "aws".	`'aws'`
`service`	`Optional[str]`	AWS service name. Defaults to "*".	`None`
`region`	`Optional[str]`	AWS region. Defaults to current region.	`None`
`account`	`Optional[str]`	AWS account ID. Defaults to current account.	`None`
`resource_id`	`Optional[str]`	Resource identifier. Defaults to "*".	`None`
`resource_type`	`Optional[str]`	Resource type prefix.	`None`
`resource_delim (Literal["/", "`		"]): Delimiter between type and ID.	required

Returns:

Type	Description
`str`	The constructed ARN string.

Source code in src/aibs_informatics_cdk_lib/common/aws/core_utils.py

def build_arn(
    partition: str = "aws",
    service: str | None = None,
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: str | None = None,
    resource_delim: Literal["/", ":"] = ":",
) -> str:
    """Build an AWS ARN string.

    Args:
        partition (str): AWS partition. Defaults to "aws".
        service (Optional[str]): AWS service name. Defaults to "*".
        region (Optional[str]): AWS region. Defaults to current region.
        account (Optional[str]): AWS account ID. Defaults to current account.
        resource_id (Optional[str]): Resource identifier. Defaults to "*".
        resource_type (Optional[str]): Resource type prefix.
        resource_delim (Literal["/", ":"]): Delimiter between type and ID.

    Returns:
        The constructed ARN string.
    """
    service = service or "*"
    region = region if region is not None else cast(str, cdk.Aws.REGION)
    account = account if account is not None else cast(str, cdk.Aws.ACCOUNT_ID)
    resource_id = resource_id or "*"

    root_arn = f"arn:{partition}:{service}:{region}:{account}"
    if resource_type is not None:
        return f"{root_arn}:{resource_type}{resource_delim}{resource_id}"
    else:
        return f"{root_arn}:{resource_id}"

build_batch_arn ¶

build_batch_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal[
        "compute-environment",
        "job",
        "job-definition",
        "job-queue",
    ]
    | None = None,
) -> str

Build an AWS Batch ARN.

Parameters:

Name	Type	Description	Default
`region`	`Optional[str]`	AWS region.	`None`
`account`	`Optional[str]`	AWS account ID.	`None`
`resource_id`	`Optional[str]`	Resource identifier.	`None`
`resource_type`	`Optional[Literal[...]]`	Batch resource type.	`None`

Returns:

Type	Description
`str`	The constructed Batch ARN string.

Source code in src/aibs_informatics_cdk_lib/common/aws/core_utils.py

def build_batch_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal["compute-environment", "job", "job-definition", "job-queue"]
    | None = None,  # noqa: E501
) -> str:
    """Build an AWS Batch ARN.

    Args:
        region (Optional[str]): AWS region.
        account (Optional[str]): AWS account ID.
        resource_id (Optional[str]): Resource identifier.
        resource_type (Optional[Literal[...]]): Batch resource type.

    Returns:
        The constructed Batch ARN string.
    """
    return build_arn(
        service="batch",
        region=region,
        account=account,
        resource_id=resource_id,
        resource_type=resource_type,
        resource_delim="/",
    )

build_dynamodb_arn ¶

build_dynamodb_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal["table"] | None = None,
) -> str

Build an AWS DynamoDB ARN.

Parameters:

Name	Type	Description	Default
`region`	`Optional[str]`	AWS region.	`None`
`account`	`Optional[str]`	AWS account ID.	`None`
`resource_id`	`Optional[str]`	Table name or resource identifier.	`None`
`resource_type`	`Optional[Literal['table']]`	DynamoDB resource type.	`None`

Returns:

Type	Description
`str`	The constructed DynamoDB ARN string.

Source code in src/aibs_informatics_cdk_lib/common/aws/core_utils.py

def build_dynamodb_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal["table"] | None = None,
) -> str:
    """Build an AWS DynamoDB ARN.

    Args:
        region (Optional[str]): AWS region.
        account (Optional[str]): AWS account ID.
        resource_id (Optional[str]): Table name or resource identifier.
        resource_type (Optional[Literal["table"]]): DynamoDB resource type.

    Returns:
        The constructed DynamoDB ARN string.
    """
    return build_arn(
        service="dynamodb",
        region=region,
        account=account,
        resource_id=resource_id,
        resource_type=resource_type,
        resource_delim="/",
    )

build_ecr_arn ¶

build_ecr_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal["repository"] | None = None,
) -> str

Build an AWS ECR ARN.

Parameters:

Name	Type	Description	Default
`region`	`Optional[str]`	AWS region.	`None`
`account`	`Optional[str]`	AWS account ID.	`None`
`resource_id`	`Optional[str]`	Repository name or resource identifier.	`None`
`resource_type`	`Optional[Literal['repository']]`	ECR resource type.	`None`

Returns:

Type	Description
`str`	The constructed ECR ARN string.

Source code in src/aibs_informatics_cdk_lib/common/aws/core_utils.py

def build_ecr_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal["repository"] | None = None,
) -> str:
    """Build an AWS ECR ARN.

    Args:
        region (Optional[str]): AWS region.
        account (Optional[str]): AWS account ID.
        resource_id (Optional[str]): Repository name or resource identifier.
        resource_type (Optional[Literal["repository"]]): ECR resource type.

    Returns:
        The constructed ECR ARN string.
    """
    return build_arn(
        service="ecr",
        region=region,
        account=account,
        resource_id=resource_id,
        resource_type=resource_type,
        resource_delim="/",
    )

build_sfn_arn ¶

build_sfn_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal[
        "*", "activity", "execution", "stateMachine"
    ]
    | None = None,
) -> str

Build an AWS Step Functions ARN.

Parameters:

Name	Type	Description	Default
`region`	`Optional[str]`	AWS region.	`None`
`account`	`Optional[str]`	AWS account ID.	`None`
`resource_id`	`Optional[str]`	State machine name or resource identifier.	`None`
`resource_type`	`Optional[Literal[...]]`	Step Functions resource type.	`None`

Returns:

Type	Description
`str`	The constructed Step Functions ARN string.

Source code in src/aibs_informatics_cdk_lib/common/aws/core_utils.py

def build_sfn_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal["*", "activity", "execution", "stateMachine"] | None = None,
) -> str:
    """Build an AWS Step Functions ARN.

    Args:
        region (Optional[str]): AWS region.
        account (Optional[str]): AWS account ID.
        resource_id (Optional[str]): State machine name or resource identifier.
        resource_type (Optional[Literal[...]]): Step Functions resource type.

    Returns:
        The constructed Step Functions ARN string.
    """
    return build_arn(
        service="states",
        region=region,
        account=account,
        resource_id=resource_id,
        resource_type=resource_type,
        resource_delim=":",
    )

build_lambda_arn ¶

build_lambda_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal[
        "function", "event-source-mapping", "layer"
    ]
    | None = None,
) -> str

Build an AWS Lambda ARN.

Parameters:

Name	Type	Description	Default
`region`	`Optional[str]`	AWS region.	`None`
`account`	`Optional[str]`	AWS account ID.	`None`
`resource_id`	`Optional[str]`	Function name or resource identifier.	`None`
`resource_type`	`Optional[Literal[...]]`	Lambda resource type.	`None`

Returns:

Type	Description
`str`	The constructed Lambda ARN string.

Source code in src/aibs_informatics_cdk_lib/common/aws/core_utils.py

def build_lambda_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal["function", "event-source-mapping", "layer"] | None = None,
) -> str:
    """Build an AWS Lambda ARN.

    Args:
        region (Optional[str]): AWS region.
        account (Optional[str]): AWS account ID.
        resource_id (Optional[str]): Function name or resource identifier.
        resource_type (Optional[Literal[...]]): Lambda resource type.

    Returns:
        The constructed Lambda ARN string.
    """
    return build_arn(
        service="lambda",
        region=region,
        account=account,
        resource_id=resource_id,
        resource_type=resource_type,
        resource_delim=":",
    )

build_s3_arn ¶

build_s3_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal[
        "bucket", "object", "accesspoint", "job"
    ]
    | None = None,
) -> str

Build an AWS S3 ARN.

Note

S3 bucket and object ARNs do not include region or account.

Parameters:

Name	Type	Description	Default
`region`	`Optional[str]`	AWS region (ignored for bucket/object).	`None`
`account`	`Optional[str]`	AWS account ID (ignored for bucket/object).	`None`
`resource_id`	`Optional[str]`	Bucket name or object path.	`None`
`resource_type`	`Optional[Literal[...]]`	S3 resource type.	`None`

Returns:

Type	Description
`str`	The constructed S3 ARN string.

Source code in src/aibs_informatics_cdk_lib/common/aws/core_utils.py

def build_s3_arn(
    region: str | None = None,
    account: str | None = None,
    resource_id: str | None = None,
    resource_type: Literal["bucket", "object", "accesspoint", "job"] | None = None,
) -> str:
    """Build an AWS S3 ARN.

    Note:
        S3 bucket and object ARNs do not include region or account.

    Args:
        region (Optional[str]): AWS region (ignored for bucket/object).
        account (Optional[str]): AWS account ID (ignored for bucket/object).
        resource_id (Optional[str]): Bucket name or object path.
        resource_type (Optional[Literal[...]]): S3 resource type.

    Returns:
        The constructed S3 ARN string.
    """
    # https://docs.aws.amazon.com/AmazonS3/latest/userguide/list_amazons3.html#amazons3-resources-for-iam-policies
    # See table above to see why resource type is set to None
    if resource_type in ["bucket", "object"]:
        resource_type = None
        # ARNs for buckets and objects CANNOT have REGION information
        region = ""
        account = ""

    return build_arn(
        service="s3",
        region=region,
        account=account,
        resource_id=resource_id,
        resource_type=resource_type,
        resource_delim=":",
    )

AWS Utilities¶

IAM Utilities¶

iam_utils ¶

Functions¶

grant_managed_policies ¶

batch_policy_statement ¶

dynamodb_policy_statement ¶

ecs_policy_statement ¶

lambda_policy_statement ¶

s3_policy_statement ¶

secretsmanager_policy_statement ¶

ses_policy_statement ¶

sfn_policy_statement ¶

sns_policy_statement ¶

ssm_policy_statement ¶

sqs_policy_statement ¶

Step Functions Utilities¶

sfn_utils ¶

Classes¶

JsonReferencePath ¶

Attributes¶

as_key property ¶

as_reference property ¶

as_jsonpath_string property ¶

as_jsonpath_object property ¶

as_jsonpath_json_to_string property ¶

as_jsonpath_list property ¶

as_jsonpath_number property ¶

Functions¶

__new__ ¶

__add__ ¶

extend ¶

sanitize classmethod ¶

is_reference classmethod ¶

empty classmethod ¶

Core Utilities¶

core_utils ¶

Functions¶

build_arn ¶

build_batch_arn ¶

build_dynamodb_arn ¶

build_ecr_arn ¶

build_sfn_arn ¶

build_lambda_arn ¶

build_s3_arn ¶

as_key `property` ¶

as_reference `property` ¶

as_jsonpath_string `property` ¶

as_jsonpath_object `property` ¶

as_jsonpath_json_to_string `property` ¶

as_jsonpath_list `property` ¶

as_jsonpath_number `property` ¶

new ¶

add ¶

sanitize `classmethod` ¶

is_reference `classmethod` ¶

empty `classmethod` ¶