aws_stepfunction_version_and_alias#

AWS StepFunction version and alias management helper functions.

Requirements:

boto3

Optional Requirements:

boto3-stubs[lambda]

Usage example:

from aws_stepfunction_version_and_alias import (
    __version__,
    describe_state_machine,
    list_state_machine_versions,
    extract_version_from_arn,
    version_dct_to_version_int,
    get_last_published_version,
    publish_version,
    keep_n_most_recent_versions,
    describe_state_machine_alias,
    list_state_machine_aliases,
    extract_alias_from_arn,
    alias_dct_to_alias_name,
    to_route_config_dict_view,
    to_route_config_list_view,
    get_alias_routing_config,
    deploy_alias,
    delete_alias,
)

This module is originally from https://github.com/MacHu-GWU/fixa-project/blob/main/fixa/aws/aws_stepfunctions_version_and_alias.py

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.describe_state_machine(sfn_client: SFNClient, state_machine_arn: str) → Optional[dict][source]#

Describe a state machine. Return None if the state machine doesn’t exist.

Reference:

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions/client/describe_state_machine.html

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.list_state_machine_versions(sfn_client: SFNClient, state_machine_arn: str, max_results: int = 1000) → List[dict][source]#

List all state machine versions. Return a list of detail dict.

The AWS Official doc says that: The results are sorted in descending order of the version creation time.

Reference:

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions/client/list_state_machine_versions.html

Parameters:: sfn_client – boto3.client("stepfunctions") object.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.extract_version_from_arn(arn: str) → int[source]#

Example:

>>> extract_version_from_arn("arn:aws:states:us-east-1:123456789012:stateMachine:HelloWorld:1")
1

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.version_dct_to_version_int(versions: List[dict]) → List[int][source]#: Convert a list of state machine version detail dict to a list of version number.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.get_last_published_version(sfn_client: SFNClient, state_machine_arn: str, max_results: int = 1) → Optional[int][source]#

Get the last published version number. If there’s no published version, return None.

Parameters:: sfn_client – boto3.client("stepfunctions") object.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.publish_version(sfn_client: SFNClient, state_machine_arn: str, description: Optional[str] = None) → Tuple[bool, int][source]#

Publish a new version. The AWS official doc says that: StepFunctions doesn’t publish a version if the state machine’s configuration and definition haven’t changed since the last version.

Reference:

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions/client/publish_state_machine_version.html

Parameters:: sfn_client – boto3.client("stepfunctions") object.
Returns:: a tuple of two items, first item is a boolean flag to indicate that if a new version is created. the second item is the version id. if there’s a new version is created, return the new version, otherwise, return the latest version number.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.keep_n_most_recent_versions(sfn_client: SFNClient, state_machine_arn: str, n: int, max_items: int = 1000, skip_in_use_version: bool = True) → List[int][source]#

Only keep the most recent n versions, delete the rest of published versions. If a version is associated with an alias, it will not be deleted.

Parameters:

sfn_client – boto3.client("stepfunctions") object.
n – number of latest version to keep
max_items – max number of versions to list in one request.
skip_in_use_version – if True, skip the version that is in use. if False, raise an exception if the version is in use.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.describe_state_machine_alias(sfn_client: SFNClient, state_machine_alias_arn: str) → Optional[dict][source]#

Describe a state machine alias. Return None if the alias doesn’t exist.

Reference:

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions/client/describe_state_machine_alias.html

Parameters:: sfn_client – boto3.client("stepfunctions") object.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.list_state_machine_aliases(sfn_client: SFNClient, state_machine_arn: str, max_results: int = 1000) → List[dict][source]#

List all state machine versions. Return a list of detail dict.

The AWS Official doc says that: The results are sorted in descending order of the version creation time.

Reference:

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions/client/list_state_machine_aliases.html

Parameters:: sfn_client – boto3.client("stepfunctions") object.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.extract_alias_from_arn(arn: str) → int[source]#

Example:

>>> extract_alias_from_arn("arn:aws:states:us-east-1:123456789012:stateMachine:HelloWorld:LIVE")
'LIVE'

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.alias_dct_to_alias_name(alias_list: List[dict]) → List[int][source]#: Convert a list of state machine alias detail dict to a list of alias name.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.to_route_config_dict_view(route_config_list_view: List[dict]) → Dict[str, int][source]#

AWS API returns the route configuration as a list of dict:

[
    {
        'stateMachineVersionArn': 'string',
        'weight': 123
    },
]

We would like to convert it into a dict to make it easier to compare.

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.get_alias_routing_config(sfn_client: SFNClient, state_machine_alias_arn: str) → Dict[int, int][source]#

The original describe_state_machine_alias API returns in this format:

{
    ...,
    'routingConfiguration': [
        {
            'stateMachineVersionArn': 'arn:...:name:2',
            'weight': 80
        },
        {
            'stateMachineVersionArn': 'arn:...:name:1',
            'weight': 20
        },
    ],
    ...
}

We would like to convert into a dict to show the version and weight:

{2: 80, 1: 20}

Parameters:: sfn_client – boto3.client("stepfunctions") object

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.deploy_alias(sfn_client: SFNClient, state_machine_arn: str, alias: str, description: Optional[str] = None, version1: Optional[Union[str, int]] = None, version2: Optional[Union[str, int]] = None, version2_percentage: Optional[int] = None, delay: int = 0) → Tuple[bool, Optional[Dict[int, int]]][source]#

Point the alias to the given version or split traffic between two versions.

Parameters:

sfn_client – boto3.client("stepfunctions") object
state_machine_arn – state machine name
alias – alias name
description – description of the alias
version1 – the main version of the alias; if not specified, use $LATEST
version2 – the secondary version of the alias; if not specified, then the version1 will have 100% traffic; if specified, then version2_percentage also has to be specified.
version2_percentage – if version2 is specified, then it has to be a value between 1 - 99.
delay – the delay in seconds to wait before doing any deployment, because your previous deployment may take time to finish.

Returns:

a tuple of two items; first item is a boolean flag to indicate whether a creation or update is performed; second item is the alias revision id, if creation or update is not performed, then return None.

Reference:

aws_ops_alpha.vendor.aws_stepfunction_version_and_alias.delete_alias(sfn_client: SFNClient, state_machine_alias_arn: str) → Optional[dict][source]#

An idempotent version of the original API.

Parameters:: sfn_client – boto3.client("stepfunctions") object