Я написал простую обертку в AWS DynamoDB. Он охватывает все операции DynamoDB. Пожалуйста, дайте звезду для этого репозитория, если вы найдете это полезным. Ссылка: https://github.com/dineshsonachalam/lucid-dynamodb
Монтаж
pip install LucidDynamodb
Примечание: Предварительное условие для разработки Python3
Оглавление
- Пример
- Создать новую таблицу
- Получите все имена таблиц
- Создать новый элемент
- Прочтите элемент
- Увеличьте существующее значение атрибута
- Обновить существующий атрибут в элементе
- Добавить новый атрибут в элемент
- Добавить атрибут в список
- Добавьте атрибут в набор строки
- Удалить атрибут из набора строки
- Удалить атрибут из элемента
- Читать элементы по фильтру
- Удалить таблицу
Пример
1. Создать новую таблицу
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")))
Выход:
INFO:root:dev_jobs table created successfully
2. Получите все имена таблиц
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))
Выход:
INFO:root:Table names: ['dev_jobs', 'user']
3. Создать новый элемент
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")
Выход:
INFO:root:Item created successfully
4. Прочтите элемент
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")
Выход:
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. Увеличьте существующее значение атрибута
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")
Выход:
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. Обновить существующий атрибут в элементе
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")
Выход:
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. Добавить новый атрибут в элемент
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")
Выход:
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. Добавить атрибут в список
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")
Выход:
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. Добавьте атрибут в набор строки
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")
Выход:
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. Удалить атрибут из набора строки
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")
Выход:
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. Удалить атрибут из элемента
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")
Выход:
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. Читать элементы по фильтру
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")
Выход:
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. Удалить таблицу
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))
Выход:
INFO:root: Table deleted successfully INFO:root:Table names: ['user']
Лицензия
Этот проект лицензирован в соответствии с условиями лицензии MIT.
Оригинал: “https://dev.to/dineshsonachalam/a-minimalistic-python-wrapper-to-aws-dynamodb-4pm0”