4
Github GitHub - dineshsonachalam/Lucid-Dynamodb: A simple Python wrapper to AWS...
source link: https://github.com/dineshsonachalam/Lucid-Dynamodb
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.
A minimalistic wrapper to AWS DynamoDB
Installation
pip install LucidDynamodb
Note: Prerequisite for Python3 development
Table of Contents
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.
Recommend
About Joyk
Aggregate valuable and interesting links.
Joyk means Joy of geeK