https://github.com/dineshsonachalam/Lucid-Dynamodb

Skip to content
 
Sign up

  * Why GitHub?
    Features -
      + Mobile -
      + Actions -
      + Codespaces -
      + Packages -
      + Security -
      + Code review -
      + Project management -
      + Integrations -
      + GitHub Sponsors -
      + Customer stories-
  * Team
  * Enterprise
  * Explore
      + Explore GitHub -

    Learn and contribute

      + Topics -
      + Collections -
      + Trending -
      + Learning Lab -
      + Open source guides -

    Connect with others

      + The ReadME Project -
      + Events -
      + Community forum -
      + GitHub Education -
      + GitHub Stars program -
  * Marketplace
  * Pricing
    Plans -
      + Compare plans -
      + Contact Sales -
      + Education -

[                    ] [search-key]

  *  
    #
    In this repository All GitHub |
    Jump to |

  * No suggested jump to results

  *  
    #
    In this repository All GitHub |
    Jump to |
  *  
    #
    In this user All GitHub |
    Jump to |
  *  
    #
    In this repository All GitHub |
    Jump to |

Sign in Sign up
{{ message }}

dineshsonachalam / Lucid-Dynamodb

  * Notifications
  * Star 43
  * Fork 2

A simple Python wrapper to AWS Dynamodb.

MIT License
43 stars 2 forks
Star
Notifications

  * Code
  * Issues 0
  * Pull requests 0
  * Actions
  * Projects 0
  * Wiki
  * Security
  * Insights

More

  * Code
  * Issues
  * Pull requests
  * Actions
  * Projects
  * Wiki
  * Security
  * Insights

master
Switch branches/tags
[                    ]
Branches Tags
Could not load branches
Nothing to show
{{ refName }} default View all branches
Could not load tags
Nothing to show
{{ refName }} default
View all tags
2 branches 0 tags
Code
 
Clone
HTTPS GitHub CLI
[https://github.com/d]

Use Git or checkout with SVN using the web URL.

[gh repo clone dinesh]

Work fast with our official CLI. Learn more.

  * Open with GitHub Desktop
  * Download ZIP

Launching GitHub Desktop

If nothing happens, download GitHub Desktop and try again.

Go back

Launching GitHub Desktop

If nothing happens, download GitHub Desktop and try again.

Go back

Launching Xcode

If nothing happens, download Xcode and try again.

Go back

Launching Visual Studio Code

Your codespace will open once ready.

There was a problem preparing your codespace, please try again.

Latest commit

@dineshsonachalam
dineshsonachalam code cleanup
...
ea1fe4f May 30, 2021
code cleanup
ea1fe4f

Git stats

  * 82 commits

Files

Permalink
Failed to load latest commit information.
Type
Name
Latest commit message
Commit time
.github/workflows
code cleanup
May 30, 2021
LucidDynamodb
Added integration test
May 30, 2021
tests
Added integration test
May 30, 2021
.gitignore
Updated documentation
May 29, 2021
LICENSE
Added MIT License
May 30, 2021
README.md
code cleanup
May 30, 2021
deploy.sh
Updated CI/CD
May 25, 2021
requirements-dev.txt
code cleanup
May 30, 2021
setup.py
code cleanup
May 30, 2021
View code
Installation Table of Contents Example 1. Create a new table 2. Get
all table names 3. Create a New Item 4. Read an Item 5. Increase an
existing attribute value 6. Update existing attribute in an item 7.
Add a new attribute in an item 8. Add an attribute to the list 9. Add
an attribute to the string set 10. Delete an attribute from the
string set 11. Delete an attribute from an item 12. Read items by
filter 13. Delete a table License

README.md

                            LucidDynamodb

               A minimalistic wrapper to AWS DynamoDB

                     Deployment Package version

 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.

About

A simple Python wrapper to AWS Dynamodb.

Topics

automation dynamodb python3 boto3

Resources

Readme

License

MIT License

Releases

No releases published

Packages 0

No packages published

Languages

  * Python 98.8%
  * Shell 1.2%

  * (c) 2021 GitHub, Inc.
  * Terms
  * Privacy
  * Security
  * Status
  * Docs

 

  * Contact GitHub
  * Pricing
  * API
  * Training
  * Blog
  * About

You can't perform that action at this time.
You signed in with another tab or window. Reload to refresh your
session. You signed out in another tab or window. Reload to refresh
your session.