`pip install boto`
python
import boto.dynamodb2
from boto.dynamodb2.exceptions import ValidationException
AWS_ACCESS_KEY = 'YOUR_ACCESS_KEY'
AWS_SECRET_KEY = 'YOUR_SECRET_KEY'
def connect_to_dynamodb():
try:
connection = boto.dynamodb2.connect_to_region(
region_name='us-west-2',
aws_access_key_id=AWS_ACCESS_KEY,
aws_secret_access_key=AWS_SECRET_KEY
)
return connection
except ValidationException as e:
dynamodb_connection = connect_to_dynamodb()
python
from boto.dynamodb2.table import Table
def create_table(connection):
try:
Table.create(
'my_table',
schema=[
HashKey('user_id', data_type=NUMBER),
RangeKey('timestamp', data_type=STRING)
],
connection=connection
)
except ValidationException as e:
create_table(dynamodb_connection)
python
from boto.dynamodb2.items import Item
def insert_data(connection):
try:
my_table = Table('my_table', connection=connection)
item_data = {
'user_id': 1,
'timestamp': '2022-01-01T00:00:00Z',
'name': 'John Doe',
'age': 30
}
item = Item(my_table, data=item_data)
item.save()
except ValidationException as e:
insert_data(dynamodb_connection)
