Skip to content

Instantly share code, notes, and snippets.

@kenvontucky
Created March 2, 2020 22:33
Show Gist options
  • Save kenvontucky/f81502fab713bb5e744af12753acd906 to your computer and use it in GitHub Desktop.
Save kenvontucky/f81502fab713bb5e744af12753acd906 to your computer and use it in GitHub Desktop.
DynamoDB base
import sys
from abc import abstractmethod
import boto3
class DynamoBase:
def __init__(self, conf):
self.conf = {
'endpoint_url': 'http://localhost:8000',
'aws_access_key_id': 'dummy_key',
'aws_secret_access_key': 'dummy_secret',
'region_name': 'dummy_region'
}
try:
self.dynamodb = boto3.client('dynamodb', **conf)
except Exception as err:
print("{} - {}".format(__name__, err))
sys.exit(1)
def create_table(self, table_schema, table_name):
self.table_name = table_name
try:
self.dynamodb.create_table(**table_schema)
except Exception as err:
print("{} - already exists - {}".format(table_name, err))
finally:
# Wait for the table to exist before exiting
waiter = self.dynamodb.get_waiter('table_exists')
waiter.wait(TableName=table_name)
def get_table(self, table_name):
dyndb = boto3.resource('dynamodb', **self.conf)
return dyndb.Table(table_name)
def list_all(self):
"""
For TESTING puposes ONLY, should not be used in
production
"""
return self.dynamodb.scan(TableName=self.table_name)
@abstractmethod
def get_params(self, key):
pass
@abstractmethod
def put_params(self, key, data):
pass
@abstractmethod
def update_params(self, key, data):
pass
@abstractmethod
def remove_params(self, key, data):
pass
def get(self, key):
"""
Get from DynamoDB
"""
params = self.get_params(key)
response = self.dynamodb.get_item(**params)
return response
def put(self, key, data):
"""
Write to DynamoDB
"""
params = self.put_params(key, data)
self.dynamodb.put_item(**params)
def update(self, key, data):
"""
Update to DynamoDB
"""
params = self.update_params(key, data)
self.dynamodb.update_item(**params)
def exists(self, key):
"""
Returns a boolean depending
on the existence of the key
"""
data = self.get(key)
return True if 'Item' in data else False
def remove(self, key):
"""
Removes a key
"""
params = self.remove_params(key)
self.dynamodb.delete_item(**params)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment