Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_utc_date_time_serialize(self):
"""
UTCDateTimeAttribute.serialize
"""
tstamp = datetime.now()
attr = UTCDateTimeAttribute()
assert attr.serialize(tstamp) == tstamp.replace(tzinfo=UTC).strftime(DATETIME_FORMAT)
host = "http://localhost:8000"
projection = AllProjection()
winner_id = UnicodeAttribute(hash_key=True)
created_time = UnicodeAttribute(range_key=True)
class GameModel(Model):
class Meta:
read_capacity_units = 1
write_capacity_units = 1
table_name = "GameModel"
host = "http://localhost:8000"
player_id = UnicodeAttribute(hash_key=True)
created_time = UTCDateTimeAttribute(range_key=True)
winner_id = UnicodeAttribute()
loser_id = UnicodeAttribute(null=True)
player_opponent_index = GamePlayerOpponentIndex()
opponent_time_index = GameOpponentTimeIndex()
class OldStyleModel(Model):
_table_name = 'IndexedModel'
user_name = UnicodeAttribute(hash_key=True)
class EmailIndex(GlobalSecondaryIndex):
"""
A global secondary index for email addresses
"""
def test_utc_date_time_deserialize(self):
"""
UTCDateTimeAttribute.deserialize
"""
tstamp = datetime.now(UTC)
attr = UTCDateTimeAttribute()
assert attr.deserialize(tstamp.strftime(DATETIME_FORMAT)) == tstamp
@convert_pynamo_attribute.register(attributes.UTCDateTimeAttribute)
def convert_date_to_string(type, attribute, registry=None):
return String(description=getattr(attribute, 'attr_name'),
required=not (getattr(attribute, 'null', True)))
index_name = "github-user-external-id-index"
write_capacity_units = int(cla.conf["DYNAMO_WRITE_UNITS"])
read_capacity_units = int(cla.conf["DYNAMO_READ_UNITS"])
projection = AllProjection()
user_external_id = UnicodeAttribute(hash_key=True)
class BaseModel(Model):
"""
Base pynamodb model used for all CLA models.
"""
date_created = UTCDateTimeAttribute(default=datetime.datetime.now())
date_modified = UTCDateTimeAttribute(default=datetime.datetime.now())
version = UnicodeAttribute(default="v1") # Schema version.
def __iter__(self):
"""Used to convert model to dict for JSON-serialized string."""
for name, attr in self.get_attributes().items():
if isinstance(attr, ListAttribute):
if attr is None or getattr(self, name) is None:
yield name, None
else:
values = attr.serialize(getattr(self, name))
if len(values) < 1:
yield name, []
else:
key = list(values[0].keys())[0]
yield name, [value[key] for value in values]
else:
class Cajero(Model):
class Meta:
table_name = 'ubicajeros_cajeros'
host = dynamodb_host
address = UnicodeAttribute()
id = UnicodeAttribute(hash_key=True)
lat = NumberAttribute(null=True)
lon = NumberAttribute(null=True)
open_hours = UnicodeAttribute(null=True)
org_code = NumberAttribute(null=True)
org_name = UnicodeAttribute(null=True)
state = UnicodeAttribute(null=True)
updated_at = UTCDateTimeAttribute(default=datetime.utcnow())
zip_code = UnicodeAttribute(null=True)
def to_dict(self):
return self.__dict__['attribute_values']
log.propagate = True
class Thread(Model):
class Meta:
read_capacity_units = 1
write_capacity_units = 1
table_name = "Thread"
host = "http://localhost:8000"
forum_name = UnicodeAttribute(hash_key=True)
subject = UnicodeAttribute(range_key=True)
views = NumberAttribute(default=0)
replies = NumberAttribute(default=0)
answered = NumberAttribute(default=0)
tags = UnicodeSetAttribute()
last_post_datetime = UTCDateTimeAttribute(null=True)
# Delete the table
# print(Thread.delete_table())
# Create the table
if not Thread.exists():
Thread.create_table(wait=True)
# Create a thread
thread_item = Thread(
'Some Forum',
'Some Subject',
tags=['foo', 'bar'],
last_post_datetime=datetime.now()
)
class AssetModel(Model):
class Meta:
table_name = os.environ['DYNAMODB_TABLE']
if 'ENV' in os.environ:
host = 'http://localhost:8000'
else:
region = os.environ['REGION']
host = os.environ['DYNAMODB_HOST']
# 'https://dynamodb.us-east-1.amazonaws.com'
asset_id = UnicodeAttribute(hash_key=True)
state = UnicodeAttribute(null=False, default=State.CREATED.name)
createdAt = UTCDateTimeAttribute(null=False, default=datetime.now().astimezone())
updatedAt = UTCDateTimeAttribute(null=False, default=datetime.now().astimezone())
def __str__(self):
return 'asset_id:{}, state:{}'.format(self.asset_id, self.state)
def get_key(self):
return u'{}/{}'.format(KEY_BASE, self.asset_id)
def save(self, conditional_operator=None, **expected_values):
try:
self.updatedAt = datetime.now().astimezone()
logger.debug('saving: {}'.format(self))
super(AssetModel, self).save()
except Exception as e:
logger.error('save {} failed: {}'.format(self.asset_id, e), exc_info=True)
raise e
"""
A DynamoDB Server Host.
"""
class Meta:
table_name = settings.value.DYNAMODB_TABLE_HOSTS
if settings.value.APPLICATION_ENV == 'development':
host = settings.value.DYNAMODB_URL
session_cls = CustomPynamoSession
service = UnicodeAttribute(hash_key=True)
ip_address = UnicodeAttribute(range_key=True)
service_repo_name_index = ServiceRepoNameIndex()
service_repo_name = UnicodeAttribute(null=True)
port = NumberAttribute()
last_check_in = UTCDateTimeAttribute()
revision = UnicodeAttribute()
tags = JSONAttribute()
class ShoppingQuery(MapAttribute):
# ref: https://developers.naver.com/docs/search/shopping/
query = UnicodeAttribute()
display = NumberAttribute(default=50)
sort = UnicodeAttribute(default='sim')
class PriceRecord(Model):
class Meta:
table_name = "price_fairy_price_record"
region = 'ap-northeast-2'
product_id = UnicodeAttribute(hash_key=True)
crawled_at = UTCDateTimeAttribute(range_key=True)
item = MapAttribute()
class Product(Model):
class Meta:
table_name = "price_fairy_product"
region = 'ap-northeast-2'
id = UnicodeAttribute(hash_key=True)
do_crawl = BooleanAttribute(default=False)
created_at = UnicodeAttribute()
last_crawled_at = UTCDateTimeAttribute(null=True)
min_price = NumberAttribute(default=0) # min_price 가격 이상인 결과만 사용
queries = ListAttribute(of=ShoppingQuery)