A minimalistic wrapper to AWS DynamoDB
Useful links:
- See the full documentation at https://lucid-dynamodb.dineshsonachalam.com
- Ask questions in the GitHub issues
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']
"""
Running Tests
To run tests, run the following command
pytest -s
License
1 maintainer
Metadata
1 maintainer
Release files for LucidDynamodb 1.0.128
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | Actions |
|---|---|---|---|---|
| LucidDynamodb-1.0.128.tar.gz | 16.0 kB | Without provenance | Details for LucidDynamodb-1.0.128.tar.gz |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Actions Reset |
|---|---|---|---|---|
| LucidDynamodb-1.0.128-py3-none-any.whl | Python 3 | none | any | Details for LucidDynamodb-1.0.128-py3-none-any.whl |
Total release size: 28.0 kB
Release files / LucidDynamodb-1.0.128.tar.gz
| Download URL | LucidDynamodb-1.0.128.tar.gz |
|---|---|
| Size | 16.0 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
d0af6c9453ccbafcf79a90e46e3d73d0a160038d212a8b839ebc766919e21a02
|
|
BLAKE2b-256 checksum How to use checksums |
6bbf55b8ebf5d8737113426dd0dfb4fd1d019d304a76a0fc78109fae9bf92b28
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/3.4.1 importlib_metadata/4.6.3 pkginfo/1.7.1 requests/2.25.1 requests-toolbelt/0.9.1 tqdm/4.62.0 CPython/3.8.10
|
Release files / LucidDynamodb-1.0.128-py3-none-any.whl
| Download URL | LucidDynamodb-1.0.128-py3-none-any.whl |
|---|---|
| Size | 12.0 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
8f63b6b12f46e0fa58682ba1540d92a3ebea695a4be0d96e41ee833871cbcdca
|
|
BLAKE2b-256 checksum How to use checksums |
1deee792873f616c8a8fdb6b43f3217e2e209896365388bcc275c87aa545a789
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/3.4.1 importlib_metadata/4.6.3 pkginfo/1.7.1 requests/2.25.1 requests-toolbelt/0.9.1 tqdm/4.62.0 CPython/3.8.10
|
Security for LucidDynamodb 1.0.128
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.
