types-aiobotocore-stepfunctions

import asyncio import os from typing import TYPE_CHECKING from aiobotocore.session import get_session # These imports are only for type checking and will not be present at runtime if TYPE_CHECKING: from types_aiobotocore_stepfunctions.client import SFNClient from types_aiobotocore_stepfunctions.type_defs import StartExecutionOutputTypeDef async def start_step_function_execution(state_machine_arn: str, execution_name: str, input_data: str): session = get_session() async with session.create_client( "stepfunctions", region_name=os.environ.get('AWS_REGION', 'us-east-1'), aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', 'DUMMY_KEY'), aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', 'DUMMY_SECRET'), ) as client: # client is implicitly typed as SFNClient by mypy/pyright after installing types-aiobotocore-stepfunctions if TYPE_CHECKING: client: SFNClient print(f"Starting execution '{execution_name}' for state machine '{state_machine_arn}'") response: StartExecutionOutputTypeDef = await client.start_execution( stateMachineArn=state_machine_arn, name=execution_name, input=input_data, ) print(f"Execution ARN: {response['executionArn']}") print(f"Start Date: {response['startDate']}") if __name__ == "__main__": # Replace with your actual State Machine ARN and desired execution details STATE_MACHINE_ARN = os.environ.get('STATE_MACHINE_ARN', 'arn:aws:states:us-east-1:123456789012:stateMachine:MyStateMachine') EXECUTION_NAME = os.environ.get('EXECUTION_NAME', 'my-test-execution-123') INPUT_DATA = os.environ.get('INPUT_DATA', '{"key": "value"}') asyncio.run(start_step_function_execution(STATE_MACHINE_ARN, EXECUTION_NAME, INPUT_DATA))