A minimalistic Python wrapper to AWS Dynamodb
Dinesh
Posted on June 1, 2021
I wrote a simple wrapper to AWS DynamoDB. It covers all the DynamoDB operations. Please give a star for this repository, if you find it useful.
Reference: https://github.com/dineshsonachalam/Lucid-Dynamodb
Installation
pip install LucidDynamodb
Note: Prerequisite for Python3 development
Table of Contents
-
Example
- Create a new table
- Get all table names
- Create a New Item
- Read an Item
- Increase an existing attribute value
- Update existing attribute in an item
- Add a new attribute in an item
- Add an attribute to the list
- Add an attribute to the string set
- Delete an attribute from the string set
- Delete an attribute from an item
- Read items by filter
- Delete a table
Example
1. Create a new table
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
table_schema = {
"TableName": "dev_jobs",
"KeySchema": [
{
"AttributeName": "company_name",
"KeyType": "HASH"
},
{
"AttributeName": "role_id",
"KeyType": "RANGE"
}
],
"AttributeDefinitions": [
{
"AttributeName": "company_name",
"AttributeType": "S"
},
{
"AttributeName": "role_id",
"AttributeType": "S"
}
],
"GlobalSecondaryIndexes": [],
"ProvisionedThroughput": {
"ReadCapacityUnits": 1,
"WriteCapacityUnits": 1
}
}
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
table_creation_status = db.create_table(
TableName=table_schema.get("TableName"),
KeySchema=table_schema.get("KeySchema"),
AttributeDefinitions=table_schema.get("AttributeDefinitions"),
GlobalSecondaryIndexes=table_schema.get("GlobalSecondaryIndexes"),
ProvisionedThroughput=table_schema.get("ProvisionedThroughput")
)
if(table_creation_status == True):
logging.info("{} table created successfully".format(table_schema.get("TableName")))
else:
logging.error("{} table creation failed".format(table_schema.get("TableName")))
Output:
INFO:root:dev_jobs table created successfully
2. Get all table names
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
table_names = db.read_all_table_names()
logging.info("Table names: {}".format(table_names))
Output:
INFO:root:Table names: ['dev_jobs', 'user']
3. Create a New Item
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
item_creation_status = db.create_item(
TableName="dev_jobs",
Item={
"company_name": "Google",
"role_id": str(uuid.uuid4()),
"role": "Software Engineer 1",
"salary": "$1,50,531",
"locations": ["Mountain View, California", "Austin, Texas", "Chicago, IL"],
"yearly_hike_percent": 8,
"benefits": set(["Internet, Medical, Edu reimbursements",
"Health insurance",
"Travel reimbursements"
]),
"overall_review":{
"overall_rating" : "4/5",
"compensation_and_benefits": "3.9/5"
}
}
)
if(item_creation_status == True):
logging.info("Item created successfully")
else:
logging.warning("Item creation failed")
Output:
INFO:root:Item created successfully
4. Read an Item
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
item = db.read_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
})
if(item != None):
logging.info("Item: {}".format(item))
else:
logging.warning("Item doesn't exist")
Output:
INFO:root:Item:
{
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5'
},
'company_name': 'Google',
'role': 'Software Engineer 1',
'yearly_hike_percent': Decimal('8'),
'salary': '$1,50,531',
'benefits': {
'Health insurance',
'Travel reimbursements',
'Internet, Medical, Edu reimbursements'
}
}
5. Increase an existing attribute value
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
increase_attribute_status = db.increase_attribute_value(
TableName='dev_jobs',
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
},
AttributeName="yearly_hike_percent",
IncrementValue=5
)
if(increase_attribute_status==True):
logging.info("Attribute value increment completed")
else:
logging.warning("Attribute value increment failed")
item = db.read_item(
TableName='dev_jobs',
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
})
if(item != None):
logging.info("Item: {}".format(item))
else:
logging.warning("Item doesn't exist")
Output:
INFO:root: Attribute value increment completed
INFO:root: Item:
{
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5'
},
'company_name': 'Google',
'role': 'Software Engineer 1',
'yearly_hike_percent': Decimal('13'),
'salary': '$1,50,531',
'benefits': {
'Health insurance',
'Travel reimbursements',
'Internet, Medical, Edu reimbursements'
}
}
6. Update existing attribute in an item
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
item_update_status = db.update_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
},
AttributesToUpdate={
'role': 'Staff Software Engineer 2'
}
)
if(item_update_status == True):
logging.info("Update is successful")
else:
logging.warning("Update failed")
item = db.read_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
})
if(item != None):
logging.info("Item: {}".format(item))
else:
logging.warning("Item doesn't exist")
Output:
INFO:root:Update is successful
INFO:root:Item:
{
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5'
},
'company_name': 'Google',
'role': 'Staff Software Engineer 2',
'yearly_hike_percent': Decimal('13'),
'salary': '$1,50,531',
'benefits': {
'Internet, Medical, Edu reimbursements',
'Travel reimbursements',
'Health insurance'
}
}
7. Add a new attribute in an item
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
item_update_status = db.update_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
},
AttributesToUpdate={
'overall_review.yearly_bonus_percent': 12
}
)
if(item_update_status == True):
logging.info("Update is successful")
else:
logging.warning("Update failed")
item = db.read_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
})
if(item != None):
logging.info("Item: {}".format(item))
else:
logging.warning("Item doesn't exist")
Output:
INFO:root:Update is successful
INFO:root:Item:
{
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5',
'yearly_bonus_percent': Decimal('12')
},
'company_name': 'Google',
'role': 'Staff Software Engineer 2',
'yearly_hike_percent': Decimal('13'),
'salary': '$1,50,531',
'benefits': {
'Internet, Medical, Edu reimbursements',
'Travel reimbursements',
'Health insurance'
}
}
8. Add an attribute to the list
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
item_update_status = db.update_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
},
AttributesToUpdate={
'locations': "Detroit, Michigan"
},
Operation="ADD_ATTRIBUTE_TO_LIST"
)
if(item_update_status == True):
logging.info("Update is successful")
else:
logging.warning("Update failed")
item = db.read_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
})
if(item != None):
logging.info("Item: {}".format(item))
else:
logging.warning("Item doesn't exist")
Output:
INFO:root:Update is successful
INFO:root:Item:
{
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5',
'yearly_bonus_percent': Decimal('12')
},
'company_name': 'Google',
'role': 'Staff Software Engineer 2',
'yearly_hike_percent': Decimal('13'),
'salary': '$1,50,531',
'benefits': {
'Health insurance',
'Internet, Medical, Edu reimbursements',
'Travel reimbursements'
}
}
9. Add an attribute to the string set
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
item_update_status = db.update_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
},
AttributesToUpdate={
'benefits': "Free Food"
},
Operation="ADD_ATTRIBUTE_TO_STRING_SET"
)
if(item_update_status == True):
logging.info("Update is successful")
else:
logging.warning("Update failed")
item = db.read_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
})
if(item != None):
logging.info("Item: {}".format(item))
else:
logging.warning("Item doesn't exist")
Output:
INFO:root:Update is successful
INFO:root:Item:
{
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5',
'yearly_bonus_percent': Decimal('12')
},
'company_name': 'Google',
'role': 'Staff Software Engineer 2',
'yearly_hike_percent': Decimal('13'),
'salary': '$1,50,531',
'benefits': {
'Internet, Medical, Edu reimbursements',
'Health insurance',
'Free Food',
'Travel reimbursements'
}
}
10. Delete an attribute from the string set
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
item_update_status = db.update_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
},
AttributesToUpdate={
'benefits': "Free Food"
},
Operation="DELETE_ATTRIBUTE_FROM_STRING_SET"
)
if(item_update_status == True):
logging.info("Update is successful")
else:
logging.warning("Update failed")
item = db.read_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
})
if(item != None):
logging.info("Item: {}".format(item))
else:
logging.warning("Item doesn't exist")
Output:
INFO:root:Update is successful
INFO:root:Item:
{
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5',
'yearly_bonus_percent': Decimal('12')
},
'company_name': 'Google',
'role': 'Staff Software Engineer 2',
'yearly_hike_percent': Decimal('13'),
'salary': '$1,50,531',
'benefits': {
'Health insurance',
'Internet, Medical, Edu reimbursements',
'Travel reimbursements'
}
}
11. Delete an attribute from an item
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
attribute_delete_status = db.delete_attribute(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
},
AttributeName="yearly_hike_percent")
if(attribute_delete_status == True):
logging.info("The attribute is deleted successfully")
else:
logging.warning("The attribute delete operation failed")
item = db.read_item(
TableName="dev_jobs",
Key={
"company_name": "Google",
"role_id": "e85f79a7-0857-4086-afbd-da13ec76b442"
})
if(item != None):
logging.info("Item: {}".format(item))
else:
logging.warning("Item doesn't exist")
Output:
INFO:root:The attribute is deleted successfully
INFO:root:Item:
{
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5',
'yearly_bonus_percent': Decimal('12')
},
'company_name': 'Google',
'role': 'Staff Software Engineer 2',
'salary': '$1,50,531',
'benefits': {
'Internet, Medical, Edu reimbursements',
'Travel reimbursements',
'Health insurance'
}
}
12. Read items by filter
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
item_creation_status = db.create_item(
TableName="dev_jobs",
Item={
"company_name": "Google",
"role_id": str(uuid.uuid4()),
"role": "Software Architect",
"salary": "$4,80,000",
"locations": ["Mountain View, California"],
"yearly_hike_percent": 13,
"benefits": set(["Internet reimbursements"]),
"overall_review":{
"overall_rating" : "3/5",
"compensation_and_benefits": "4.2/5"
}
}
)
if(item_creation_status == True):
logging.info("Item created successfully")
else:
logging.warning("Item creation failed")
items = db.read_items_by_filter(
TableName='dev_jobs',
KeyConditionExpression=Key("company_name").eq("Google")
)
if(len(items)>0):
logging.info("Items: {}".format(items))
else:
logging.warning("Items doesn't exist")
Output:
INFO:root: Item created successfully
INFO:root:Items:
[{
'locations': ['Mountain View, California'],
'role_id': 'b6065b19-4333-43a4-abf7-dedca2880669',
'overall_review': {
'compensation_and_benefits': '4.2/5',
'overall_rating': '3/5'
},
'company_name': 'Google',
'role': 'Software Architect',
'yearly_hike_percent': Decimal('13'),
'salary': '$4,80,000',
'benefits': {
'Internet reimbursements'
}
}, {
'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
'role_id': 'e85f79a7-0857-4086-afbd-da13ec76b442',
'overall_review': {
'compensation_and_benefits': '3.9/5',
'overall_rating': '4/5',
'yearly_bonus_percent': Decimal('12')
},
'company_name': 'Google',
'role': 'Staff Software Engineer 2',
'salary': '$1,50,531',
'benefits': {
'Health insurance',
'Internet, Medical, Edu reimbursements',
'Travel reimbursements'
}
}]
13. Delete a table
from LucidDynamodb.Operations import DynamoDb
import os
import logging
import uuid
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
if __name__ == "__main__":
db = DynamoDb(region_name="us-east-1",
aws_access_key_id=AWS_ACCESS_KEY_ID,
aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
delete_table_status = db.delete_table(TableName='dev_jobs')
if(delete_table_status == True):
logging.info("Table deleted successfully")
else:
logging.warning("Table delete operation failed")
table_names = db.read_all_table_names()
logging.info("Table names: {}".format(table_names))
Output:
INFO:root: Table deleted successfully
INFO:root:Table names: ['user']
License
This project is licensed under the terms of the MIT license.
💖 💪 🙅 🚩
Dinesh
Posted on June 1, 2021
Join Our Newsletter. No Spam, Only the good stuff.
Sign up to receive the latest update from our blog.