
This assumes you are familar with the basic configurations needed to connect to Amazon Web Services (AWS) using Python boto3. If not, check out my article Python (Scripting) - Getting Started with Amazon Web Services (AWS) boto3.
This also assumes you have created one or more Step Function State Machines.
Here is the minimal boilerplate code without any error handling to list your EC2 instances. This will use the default profile in your .aws/credentials profile. If you want to used a named profile, check out my article Python (Scripting) - Getting Started with Amazon Web Services (AWS) boto3.
import boto3
import os
client = boto3.client('ec2')
response = client.describe_instances(
Filters = [
{ 'Name': 'tag:environment', 'Values': ['staging'] },
{ 'Name': 'tag:purpose', 'Values': ['postgres'] }
]
)
print(f"response = {response}")
Which should return something like this.
{'stateMachines': [
{'stateMachineArn': 'arn:aws:states:us-east-1:123456789012:stateMachine:myStateMachine',
'name': 'myStateMachine',
'type': 'STANDARD',
'creationDate': datetime.datetime(2023, 12, 27, 2, 39, 32, 497000, tzinfo=tzlocal())}
],
'ResponseMetadata': {
'RequestId': 'dcac2e33-effa-4f60-95c3-d7bbba7f1351',
'HTTPStatusCode': 200,
'HTTPHeaders': {
'x-amzn-requestid': 'dcac2e33-effa-4f60-95c3-d7bbba7f1351',
'date': 'Thu, 28 Dec 2023 12:07:48 GMT',
'content-type': 'application/x-amz-json-1.0',
'content-length': '211',
'connection': 'keep-alive'},
'RetryAttempts': 0
}
}
The "default" profile in your .aws/credentials file will be used. Session can be used to use some other profile.
#!/usr/bin/python3
import boto3
session = boto3.Session(profile_name='johndoe')
client = session.client('ec2')
response = client.describe_instances(
Filters = [
{ 'Name': 'tag:environment', 'Values': ['staging'] },
{ 'Name': 'tag:purpose', 'Values': ['postgres'] }
]
)
print(f"response = {response}")
Or, os.environ['AWS_PROFILE'] can be used to specify the profile in /home/john.doe/.aws/config and /home/john.doe/.aws/credentials to use.
import boto3
import os
os.environ['AWS_PROFILE'] = 'johndoe'
client = boto3.client('ec2')
response = client.describe_instances(
Filters = [
{ 'Name': 'tag:environment', 'Values': ['staging'] },
{ 'Name': 'tag:purpose', 'Values': ['postgres'] }
]
)
print(f"response = {response}")
And here is how you can access certain key/values from the dictionary.
#!/usr/bin/python3
import boto3
client = boto3.client('stepfunctions')
state_machines = client.list_state_machines()
for state_machine in state_machines['stateMachines']:
print(f"state machine ARN = {state_machine['stateMachineArn']}")
Which should return something like this.
state machine ARN = arn:aws:states:us-east-1:123456789012:stateMachine:myStateMachine
Now with the ARN, you can use start_execution to execute the Step Function State Machine.
#!/usr/bin/python3
import boto3
import json
import sys
client = boto3.client('stepfunctions')
state_machines = client.list_state_machines()
for state_machine in state_machines['stateMachines']:
print(f"state machine ARN = {state_machine['stateMachineArn']}")
payload = json.dumps({'email':'john.doe@example.com'})
try:
response = client.start_execution(stateMachineArn=state_machine['stateMachineArn'], input=payload)
except Exception as exception:
print(f"exception = {exception}")
else:
print(f"response = {response}")
And something like this should be returned.
{'executionArn': 'arn:aws:states:us-east-1:123456789012:execution:myStateMachine:60d888c1-a52b-461a-b5ae-b3eecba77fcf',
'startDate': datetime.datetime(2023, 12, 28, 12, 21, 14, 517000, tzinfo=tzlocal()),
'ResponseMetadata': {
'RequestId': '5bcecf56-27f6-497a-83ad-377c5b625dae',
'HTTPStatusCode': 200,
'HTTPHeaders': {
'x-amzn-requestid': '5bcecf56-27f6-497a-83ad-377c5b625dae',
'date': 'Thu, 28 Dec 2023 12:21:14 GMT',
'content-type': 'application/x-amz-json-1.0',
'content-length': '162',
'connection': 'keep-alive'
},
'RetryAttempts': 0
}
}
You will probably also want to check the execution result and any response data, perhaps like this.
#!/usr/bin/python3
import boto3
import json
import sys
client = boto3.client('stepfunctions')
state_machines = client.list_state_machines()
for state_machine in state_machines['stateMachines']:
print(f"state machine ARN = {state_machine['stateMachineArn']}")
payload = json.dumps({'email':'john.doe@example.com'})
try:
response = client.start_execution(stateMachineArn=state_machine['stateMachineArn'], input=payload)
except Exception as exception:
print(f"exception = {exception}")
else:
print(f"response = {response}")
executionArn = start_execution_response['executionArn']
attempt = 0
while attempt < 30:
try:
describe_execution = client.describe_execution(executionArn=executionArn)
except Exception as exception:
print(f"exception = {exception}")
else:
print(f"status = {describe_execution['status']}")
if describe_execution['status'] == "FAILED":
print(f"error = {describe_execution['error']}")
print(f"cause = {describe_execution['cause']}")
break
if describe_execution['status'] == "SUCCEEDED":
json_loads = json.loads(describe_execution['output'])
print(f"json_loads['ResponseBody']['result'] = {json_loads['ResponseBody']['result']}")
print(f"json_loads['ResponseBody']['message'] = {json_loads['ResponseBody']['message']}")
break
attempt += 1
time.sleep(1)
Did you find this article helpful?
If so, consider buying me a coffee over at