A simple Python wrapper to AWS Dynamodb

pip install LucidDynamodb

from LucidDynamodb import DynamoDbdb = DynamoDb()"""$ pip install awscli  #can add user flag$ aws configureAWS 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]:"""

from LucidDynamodb import DynamoDbimport osAWS_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)

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    TableAlreadyExists)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:dev_jobs table created successfully"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Table names: ['CertMagic', 'dev_jobs', 'dev_test', 'kp-config-v1', 'test-1']"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Item created successfully"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    ItemNotFound)import logginglogging.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.pyINFO: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'    }}"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    QueryFilterValidationFailed)import loggingfrom boto3.dynamodb.conditions import Keylogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Item created successfullyINFO: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'    }}]"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Update is successfulINFO: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'    }}"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Update is successfulINFO: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'    }}"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Update is successfulINFO: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'    }}"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Update is successfulINFO: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'    }}"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Attribute value increment completedINFO: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'    }}"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:The attribute is deleted successfullyINFO: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'    }}"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Update is successfulINFO: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'    }}"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    UnexpectedError)from boto3.dynamodb.conditions import Keyimport logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Item deleted successfullyINFO: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'    }}]"""

from LucidDynamodb import DynamoDbfrom LucidDynamodb.exceptions import (    TableNotFound)import logginglogging.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.pyINFO:botocore.credentials:Found credentials in environment variables.INFO:root:Table deleted successfullyINFO:root:Table names: ['CertMagic', 'dev_test', 'kp-config-v1', 'test-1']"""

pytest -s