A simple Python wrapper to AWS Dynamodb

I created a minimalistic python wrapper to AWS Dynamodb.

Github repo: https://github.com/dineshsonachalam/lucid-dynamodb

Table of contents

Installation

Example

Connect to DynamodDB
Create a new table
Get all table names
Create a New Item
Read an Item
Read items by filter
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
Increase an existing attribute value
Delete an attribute from an item
Delete an attribute from the string set
Delete an item
Delete a table

Running tests

Github Workflow Artifacts

License

Installation

pip install LucidDynamodb

Note: Prerequisite for Python3 development

Example

Connect to DynamoDB

You can connect to DynamoDB by following any of these two ways.

Using AWS config

from LucidDynamodb import DynamoDb
db = DynamoDb()

"""
$ pip install awscli  #can add user flag
$ aws configure
AWS Access Key ID [****************ABCD]:[enter your key here]
AWS Secret Access Key [****************xyz]:[enter your secret key here]
Default region name [us-west-2]:[enter your region here]
Default output format [None]:
"""

Using AWS secret key

from LucidDynamodb import DynamoDb
import os
AWS_ACCESS_KEY_ID = os.getenv("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.getenv("AWS_SECRET_ACCESS_KEY")
db = DynamoDb(region_name="us-east-1",
              aws_access_key_id=AWS_ACCESS_KEY_ID,
              aws_secret_access_key=AWS_SECRET_ACCESS_KEY)

Create a new table

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    TableAlreadyExists
)
import logging
logging.basicConfig(level=logging.INFO)

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__":
    try:
        db = DynamoDb()
        db.create_table(
            table_name=table_schema.get("TableName"),
            key_schema=table_schema.get("KeySchema"),
            attribute_definitions=table_schema.get("AttributeDefinitions"),
            global_secondary_indexes=table_schema.get("GlobalSecondaryIndexes"),
            provisioned_throughput=table_schema.get("ProvisionedThroughput")
        )
        logging.info(f"{table_schema.get('TableName')} table created successfully")
    except TableAlreadyExists as e:
        logging.error(f"{table_schema.get('TableName')} table creation failed - {e}")

"""
dineshsonachalam@macbook examples % python 1-create-a-new-table.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:dev_jobs table created successfully
"""

Get all table names

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        table_names = db.read_all_table_names()
        logging.info(f"Table names: {table_names}")
    except UnexpectedError as e:
        logging.error(f"Read all table names failed - {e}")

"""
dineshsonachalam@macbook examples % python 2-get-all-table-names.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Table names: ['CertMagic', 'dev_jobs', 'dev_test', 'kp-config-v1', 'test-1']
"""

Create a new item

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.create_item(
            table_name="dev_jobs",
            item={
                "company_name": "Google",
                "role_id": "111",
                "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"
                }
            }
        )
        logging.info("Item created successfully")
    except UnexpectedError as e:
        logging.error(f"Item creation failed - {e}")

"""
dineshsonachalam@macbook examples % python 3-create-a-new-item.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Item created successfully
"""

Read an item

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    ItemNotFound
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        item = db.read_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info(f"Item: {item}")
    except ItemNotFound as e:
        logging.error(f"Item doesn't exist - {e}")

"""
dineshsonachalam@macbook examples % python 4-read-an-item.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Item: {
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL'],
    'role_id': '111',
    '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': {
        'Travel reimbursements',
        'Internet, Medical, Edu reimbursements',
        'Health insurance'
    }
}
"""

Read items by filter

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    QueryFilterValidationFailed
)
import logging
from boto3.dynamodb.conditions import Key
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.create_item(
            table_name="dev_jobs",
            item={
                "company_name": "Google",
                "role_id": "112",
                "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"
                }
            }
        )
        logging.info("Item created successfully")
        items = db.read_items_by_filter(
                        table_name='dev_jobs',
                        key_condition_expression=Key("company_name").eq("Google")
        )
        logging.info(f"Items: {items}")
    except QueryFilterValidationFailed as e:
        logging.error(f"Items doesn't exist - {e}")

"""
dineshsonachalam@macbook examples % python 5-read-items-by-filter.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Item created successfully
INFO:root:Items: [{
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL'],
    'role_id': '111',
    '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': {
        'Internet, Medical, Edu reimbursements',
        'Travel reimbursements',
        'Health insurance'
    }
}, {
    'locations': ['Mountain View, California'],
    'role_id': '112',
    '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'
    }
}]
"""

Update existing attribute in an item

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.update_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            },
            attributes_to_update={
                'role': 'Staff Software Engineer 2'
            }
        )
        logging.info("Update is successful")
        item = db.read_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info(f"Item: {item}")
    except UnexpectedError as e:
        logging.error(f"Update failed - {e}")

"""
dineshsonachalam@macbook examples % python 6-update-existing-attribute-in-an-item.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Update is successful
INFO:root:Item: {
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL'],
    'role_id': '111',
    'overall_review': {
        'compensation_and_benefits': '3.9/5',
        'overall_rating': '4/5'
    },
    'company_name': 'Google',
    'role': 'Staff Software Engineer 2',
    'yearly_hike_percent': Decimal('8'),
    'salary': '$1,50,531',
    'benefits': {
        'Health insurance',
        'Internet, Medical, Edu reimbursements',
        'Travel reimbursements'
    }
}
"""

Add a new attribute in an item

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.update_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            },
            attributes_to_update={
                'overall_review.yearly_bonus_percent': 12
            }
        )
        logging.info("Update is successful")
        item = db.read_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info(f"Item: {item}")
    except UnexpectedError as e:
        logging.error(f"Update failed - {e}")

"""
dineshsonachalam@macbook examples % python 7-add-a-new-attribute-in-an-item.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Update is successful
INFO:root:Item: {
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL'],
    'role_id': '111',
    '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('8'),
    'salary': '$1,50,531',
    'benefits': {
        'Travel reimbursements',
        'Internet, Medical, Edu reimbursements',
        'Health insurance'
    }
}
"""

Add an attribute to the list

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.update_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            },
            attributes_to_update={
                'locations': "Detroit, Michigan"
            },
            operation="ADD_ATTRIBUTE_TO_LIST"
        )
        logging.info("Update is successful")
        item = db.read_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info(f"Item: {item}")
    except UnexpectedError as e:
        logging.error(f"Update failed - {e}")

"""
dineshsonachalam@macbook examples % python 8-add-an-attribute-to-the-list.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Update is successful
INFO:root:Item: {
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
    'role_id': '111',
    '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('8'),
    'salary': '$1,50,531',
    'benefits': {
        'Health insurance',
        'Travel reimbursements',
        'Internet, Medical, Edu reimbursements'
    }
}
"""

Add an attribute to the string set

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.update_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            },
            attributes_to_update={
                'benefits': "Free Food"
            },
            operation="ADD_ATTRIBUTE_TO_STRING_SET"
        )
        logging.info("Update is successful")
        item = db.read_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info(f"Item: {item}")
    except UnexpectedError as e:
        logging.error(f"Update failed - {e}")

"""
dineshsonachalam@macbook examples % python 9-add-an-attribute-to-the-string-set.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Update is successful
INFO:root:Item: {
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
    'role_id': '111',
    '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('8'),
    'salary': '$1,50,531',
    'benefits': {
        'Travel reimbursements',
        'Free Food',
        'Health insurance',
        'Internet, Medical, Edu reimbursements'
    }
}
"""

Increase an existing attribute value

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.increase_attribute_value(
            table_name='dev_jobs',
            key={
                "company_name": "Google",
                "role_id": "111"
            },
            attribute_name="yearly_hike_percent",
            increment_value=5
        )
        logging.info("Attribute value increment completed")
        item = db.read_item(
            table_name='dev_jobs',
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info(f"Item: {item}")
    except UnexpectedError as e:
        logging.error(f"Attribute value increment failed - {e}")

"""
dineshsonachalam@macbook examples % python 10-increase-an-existing-attribute-value.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Attribute value increment completed
INFO:root:Item: {
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
    'role_id': '111',
    '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',
        'Free Food',
        'Health insurance',
        'Travel reimbursements'
    }
}
"""

Delete an attribute from an item

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.delete_attribute(
            table_name="dev_jobs",
            key={"company_name": "Google", "role_id": "111"},
            attribute_name="yearly_hike_percent")
        logging.info("The attribute is deleted successfully")
        item = db.read_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info(f"Item: {item}")
    except UnexpectedError as e:
         logging.error(f"The attribute delete operation failed - {e}")

"""
dineshsonachalam@macbook examples % python 11-delete-an-attribute-from-an-item.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:The attribute is deleted successfully
INFO:root:Item: {
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
    'role_id': '111',
    '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': {
        'Travel reimbursements',
        'Free Food',
        'Health insurance',
        'Internet, Medical, Edu reimbursements'
    }
}
"""

Delete an attribute from the string set

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.update_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            },
            attributes_to_update={
                'benefits': "Free Food"
            },
            operation="DELETE_ATTRIBUTE_FROM_STRING_SET"
        )
        logging.info("Update is successful")
        item = db.read_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info(f"Item: {item}")
    except UnexpectedError as e:
        logging.error(f"Update failed - {e}")

"""
dineshsonachalam@macbook examples % python 12-delete-an-attribute-from-the-string-set.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Update is successful
INFO:root:Item: {
    'locations': ['Mountain View, California', 'Austin, Texas', 'Chicago, IL', 'Detroit, Michigan'],
    'role_id': '111',
    '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',
        'Health insurance',
        'Travel reimbursements'
    }
}
"""

Delete an item

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    UnexpectedError
)
from boto3.dynamodb.conditions import Key
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.delete_item(
            table_name="dev_jobs",
            key={
                "company_name": "Google",
                "role_id": "111"
            }
        )
        logging.info("Item deleted successfully")
        items = db.read_items_by_filter(
                        table_name='dev_jobs',
                        key_condition_expression=Key("company_name").eq("Google")
        )
        logging.info(f"Items: {items}") 
    except UnexpectedError as e:
        logging.warning(f"Item delete operation failed - {e}")

"""
dineshsonachalam@macbook examples % python 13-delete-an-item.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Item deleted successfully
INFO:root:Items: [{
    'locations': ['Mountain View, California'],
    'role_id': '112',
    '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'
    }
}]
"""

Delete a table

from LucidDynamodb import DynamoDb
from LucidDynamodb.exceptions import (
    TableNotFound
)
import logging
logging.basicConfig(level=logging.INFO)

if __name__ == "__main__":
    try:
        db = DynamoDb()
        db.delete_table(table_name='dev_jobs')
        logging.info("Table deleted successfully")
        table_names = db.read_all_table_names()
        logging.info(f"Table names: {table_names}")
    except TableNotFound as e:
        logging.error(f"Table delete operation failed {e}")

"""
dineshsonachalam@macbook examples % python 14-delete-a-table.py
INFO:botocore.credentials:Found credentials in environment variables.
INFO:root:Table deleted successfully
INFO:root:Table names: ['CertMagic', 'dev_test', 'kp-config-v1', 'test-1']
"""