Repository URL to install this package:
|
Version:
116.0.0 ▾
|
import logging
import datetime
import json
import hydra.settings.config
log = logging.getLogger()
def get_dynamodb_connection(region=None):
"""
:return Connection to DynamoDB for specified region
"""
from boto import dynamodb2, connect_dynamodb
if region:
return dynamodb2.connect_to_region(region)
else:
return connect_dynamodb()
def get_dynamodb_table_name(table_name):
""" construct the DynamoDB table name from prefix and with time suffixes
:param table_name: table name prefix
:return: constructed table name, ie stage_metrics_apilog_2015_04_11_00
"""
ref_date = datetime.datetime.utcnow().strftime("%Y_%m_%d_%H")
complete_name = '{0}_{1}_{2}'.format(hydra.settings.config.AWS_DYNAMODB_TABLE_PREFIX,
table_name,
ref_date)
return complete_name
def get_dynamodb_table(table_name, conn, region=None):
""" construct the DynamoDB table name from prefix and with time suffixes """
from boto.exception import JSONResponseError
from boto.dynamodb2.table import Table
# We had to use different type of connections here due some issues
# https://github.com/boto/boto/issues/2762
# with the atomic counter for the dynamodb2 classes
try:
if region:
table = Table(table_name, connection=conn)
table.describe()
else:
table = conn.get_table(table_name)
except JSONResponseError, e:
log.error("Unable to lookup DynamoDB table [%s]: %s", table_name, e)
return None
return table
class DatetimeEncoder(json.JSONEncoder):
"""
Alternate handler that can serialize dates
"""
def default(self, obj):
if hasattr(obj, 'isoformat'):
return obj.isoformat()
# Let the base class default method raise the TypeError
return json.JSONEncoder.default(self, obj)