Skip to content

Instantly share code, notes, and snippets.

@kenvontucky
Created March 2, 2020 22:37
Show Gist options
  • Save kenvontucky/f19584b9d4276ca5f4afa5ae6ef57d2f to your computer and use it in GitHub Desktop.
Save kenvontucky/f19584b9d4276ca5f4afa5ae6ef57d2f to your computer and use it in GitHub Desktop.
from dynamo.base import DynamoBase
from samples.log.schema import TABLE_NAME, TIMESTAMP, STATUS, INDEX_STATUS_KEY
class DynamoLog(DynamoBase):
def get_params(self, key):
params = {
'TableName': TABLE_NAME,
'Key': {
TIMESTAMP: {"S": key}
}
}
return params
def put_params(self, key, data):
params = {
'TableName': TABLE_NAME,
'Item': {
TIMESTAMP: {"S": key},
}
}
params['Item'].update(data)
return params
def update_params(self, timestamp, value):
params = {
'ExpressionAttributeNames': {
'#LU': INDEX_STATUS_KEY,
'#S': STATUS
},
'ExpressionAttributeValues': {
':lu': {
'S': timestamp
},
':s': {
'S': value
}
},
'Key': {
TIMESTAMP: {"S": timestamp}
},
'ReturnValues': 'UPDATED_NEW',
'TableName': TABLE_NAME,
'UpdateExpression': 'SET #LU = :lu, #S = :s'
}
return params
def filter_by_timestamp_status(self, timestamp, status):
"""
Filters all entries by timestamp and status
"""
response = self.dynamodb.query(TableName=TABLE_NAME,
KeyConditionExpression=f"{TIMESTAMP} = :updated_at",
FilterExpression="#S = :status",
ExpressionAttributeValues={":status": {"S": status},
":updated_at": {"S": timestamp}},
ExpressionAttributeNames={"#S": STATUS})
return response
def filter_by_status(self, status):
"""
Filters all entries by key two
"""
response = self.dynamodb.query(TableName=TABLE_NAME,
IndexName=INDEX_STATUS_KEY,
KeyConditionExpression="#S = :status",
ExpressionAttributeValues={":status": {"S": status}},
ExpressionAttributeNames={"#S": STATUS})
return response
def update(self, timestamp, status):
"""
Update to DynamoDB
"""
params = self.update_params(timestamp, status)
self.dynamodb.update_item(**params)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment