Create a new Dynamo table using Boto3
# Get the service resource. dynamodb = boto3.resource('dynamodb') # Create the DynamoDB table. table = dynamodb.create_table( TableName='users', KeySchema=[ { 'AttributeName': 'username', 'KeyType': 'HASH' }, { 'AttributeName': 'last_name', 'KeyType': 'RANGE' } ], AttributeDefinitions=[ { 'AttributeName': 'username', 'AttributeType': 'S' }, { 'AttributeName': 'last_name', 'AttributeType': 'S' }, ], ProvisionedThroughput={ 'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5 } ) # Wait until the table exists. table.meta.client.get_waiter('table_exists').wait(TableName='users') # Print out some data about the table. print(table.item_count)
This method will return a Dynamodb.TableOnce you have a DynamoDB.Table resource you can add new items to the table using DynamoDB.Table.put_item():
table.put_item(
Item={
'username': 'janedoe',
'first_name': 'Jane',
'last_name': 'Doe',
'age': 25,
'account_type': 'standard_user',
}
)
You can then retrieve the object using DynamoDB.Table.get_item():
response = table.get_item(
Key={
'username': 'janedoe',
'last_name': 'Doe'
}
)
item = response['Item']
print(item)
You can then update attributes of the item in the table:
table.update_item(
Key={
'username': 'janedoe',
'last_name': 'Doe'
},
UpdateExpression='SET age = :val1',
ExpressionAttributeValues={
':val1': 26
}
)
You can also delete the item using DynamoDB.Table.delete_item():
table.delete_item(
Key={
'username': 'janedoe',
'last_name': 'Doe'
}
)
With the table full of items, you can then query or scan the items in the table using the DynamoDB.Table.query() or DynamoDB.Table.scan() methods respectively. To add conditions to scanning and querying the table, you will need to import the boto3.dynamodb.conditions.Key and boto3.dynamodb.conditions.Attr classes. The boto3.dynamodb.conditions.Key should be used when the condition is related to the key of the item. The boto3.dynamodb.conditions.Attr should be used when the condition is related to an attribute of the item:
from boto3.dynamodb.conditions import Key, Attr
This queries for all of the users whose username key equals johndoe:
response = table.query(
KeyConditionExpression=Key('username').eq('johndoe')
)
items = response['Items']
print(items)
Similarly you can scan the table based on attributes of the items. For example, this scans for all the users whose age is less than 27:
response = table.scan(
FilterExpression=Attr('age').lt(27)
)
items = response['Items']
print(items)Describing instancesimport boto3
ec2 = boto3.client('ec2')response = ec2.describe_instances()
print(response)
Monitor and unmonitor instances
import sys
import boto3
ec2 = boto3.client('ec2')
if sys.argv[1] == 'ON':
response = ec2.monitor_instances(InstanceIds=['INSTANCE_ID'])
else:
response = ec2.unmonitor_instances(InstanceIds=['INSTANCE_ID'])
print(response)
Start and stop instances
import boto3
from botocore.exceptions import ClientError
instance_id = sys.argv[2]
action = sys.argv[1].upper()
ec2 = boto3.client('ec2')
if action == 'ON':
# Do a dryrun first to verify permissions
try:
ec2.start_instances(InstanceIds=[instance_id], DryRun=True)
except ClientError as e:
if 'DryRunOperation' not in str(e):
raise
# Dry run succeeded, run start_instances without dryrun
try:
response = ec2.start_instances(InstanceIds=[instance_id], DryRun=False)
print(response)
except ClientError as e:
print(e)
else:
# Do a dryrun first to verify permissions
try:
ec2.stop_instances(InstanceIds=[instance_id], DryRun=True)
except ClientError as e:
if 'DryRunOperation' not in str(e):
raise
# Dry run succeeded, call stop_instances without dryrun
try:
response = ec2.stop_instances(InstanceIds=[instance_id], DryRun=False)
print(response)
except ClientError as e:
print(e)
Reboot instances
import boto3
from botocore.exceptions import ClientError
ec2 = boto3.client('ec2')
try:
ec2.reboot_instances(InstanceIds=['INSTANCE_ID'], DryRun=True)
except ClientError as e:
if 'DryRunOperation' not in str(e):
print("You don't have permission to reboot instances.")
raise
try:
response = ec2.reboot_instances(InstanceIds=['INSTANCE_ID'], DryRun=False)
print('Success', response)
except ClientError as e:
print('Error', e)Describe Regions and Availability Zones
ec2 = boto3.client('ec2')
# Retrieves all regions/endpoints that work with EC2
response = ec2.describe_regions()
print('Regions:', response['Regions'])
# Retrieves availability zones only for region of the ec2 object
response = ec2.describe_availability_zones()
print('Availability Zones:', response['AvailabilityZones'])
Create a security group and rules
import boto3
from botocore.exceptions import ClientError
ec2 = boto3.client('ec2')
response = ec2.describe_vpcs()
vpc_id = response.get('Vpcs', [{}])[0].get('VpcId', '')
try:
response = ec2.create_security_group(GroupName='SECURITY_GROUP_NAME',
Description='DESCRIPTION',
VpcId=vpc_id)
security_group_id = response['GroupId']
print('Security Group Created %s in vpc %s.' % (security_group_id, vpc_id))
data = ec2.authorize_security_group_ingress(
GroupId=security_group_id,
IpPermissions=[
{'IpProtocol': 'tcp',
'FromPort': 80,
'ToPort': 80,
'IpRanges': [{'CidrIp': '0.0.0.0/0'}]},
{'IpProtocol': 'tcp',
'FromPort': 22,
'ToPort': 22,
'IpRanges': [{'CidrIp': '0.0.0.0/0'}]}
])
print('Ingress Successfully Set %s' % data)
except ClientError as e:
print(e)
Delete a security group
import boto3
from botocore.exceptions import ClientError
# Create EC2 client
ec2 = boto3.client('ec2')
# Delete security group
try:
response = ec2.delete_security_group(GroupId='SECURITY_GROUP_ID')
print('Security Group Deleted')
except ClientError as e:
print(e)
