DEV Community

Cover image for A simple Python wrapper to AWS Dynamodb
Dinesh
Dinesh

Posted on

A simple Python wrapper to AWS Dynamodb

#python #aws #devops #dynamodb

I created a minimalistic python wrapper to AWS Dynamodb.

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

Table of contents

Installation 

pip install LucidDynamodb
Enter fullscreen mode Exit fullscreen mode

Note: Prerequisite for Python3 development

Example

Connect to DynamoDB

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

  1. 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]:
"""
Enter fullscreen mode Exit fullscreen mode
  1. 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)
Enter fullscreen mode Exit fullscreen mode

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
"""
Enter fullscreen mode Exit fullscreen mode

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']
"""
Enter fullscreen mode Exit fullscreen mode

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
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}]
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}
"""
Enter fullscreen mode Exit fullscreen mode

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'
    }
}]
"""
Enter fullscreen mode Exit fullscreen mode

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']
"""
Enter fullscreen mode Exit fullscreen mode

Running Tests

To run tests, run the following command

pytest -s
Enter fullscreen mode Exit fullscreen mode

License

MIT © dineshsonachalam

Top comments (0)

Read next

recursivecodes profile image

Live Streaming from Unity - Broadcasting from a Meta Quest (Part 8)

Todd Sharp -

anmolbaranwal profile image

21 AI Tools that are changing the World

Anmol Baranwal -

swapnoneel123 profile image

Designing Machine Learning Workflows in Python

Swapnoneel Saha -

saranshk profile image

Node.js 20.6 adds built-in support for .env files

saransh kataria -