Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def foo(self):
return 1
class ComplexModel(Model):
class Meta:
table_name = 'ComplexModel'
person = Person(attr_name='weird_person')
key = NumberAttribute(hash_key=True)
class OfficeEmployee(Model):
class Meta:
table_name = 'OfficeEmployeeModel'
office_employee_id = NumberAttribute(hash_key=True)
person = Person()
office_location = Location()
def foo(self):
return 1
class CarInfoMap(MapAttribute):
make = UnicodeAttribute(null=False)
model = UnicodeAttribute(null=True)
class CarModel(Model):
class Meta:
table_name = 'CarModel'
car_id = NumberAttribute(hash_key=True, null=False)
else:
key = list(values[0].keys())[0]
yield name, [value[key] for value in values]
else:
yield name, attr.serialize(getattr(self, name))
class DocumentTabModel(MapAttribute):
"""
Represents a document tab in the document model.
"""
document_tab_type = UnicodeAttribute(default='text')
document_tab_id = UnicodeAttribute()
document_tab_name = UnicodeAttribute()
document_tab_page = NumberAttribute(default=1)
document_tab_position_x = NumberAttribute()
document_tab_position_y = NumberAttribute()
document_tab_width = NumberAttribute(default=200)
document_tab_height = NumberAttribute(default=20)
document_tab_is_locked = BooleanAttribute(default=False)
document_tab_is_required = BooleanAttribute(default=True)
class DocumentTab(model_interfaces.DocumentTab):
"""
ORM-agnostic wrapper for the DynamoDB DocumentTab model.
"""
def __init__(self, # pylint: disable=too-many-arguments
document_tab_type=None,
document_tab_id=None,
document_tab_name=None,
document_tab_page=None,
document_tab_position_x=None,
# For Nulls:
except AttributeError:
yield name, None
class DurableHistoricalModel(BaseHistoricalModel):
"""The base Historical Durable (Differ) Table model base class."""
eventTime = EventTimeAttribute(range_key=True, default=default_event_time)
class CurrentHistoricalModel(BaseHistoricalModel):
"""The base Historical Current Table model base class."""
eventTime = EventTimeAttribute(default=default_event_time)
ttl = NumberAttribute(default=default_ttl())
eventSource = UnicodeAttribute()
class AWSHistoricalMixin(BaseHistoricalModel):
"""This is the main Historical event mixin. All the major required (and optional) fields are here."""
arn = UnicodeAttribute(hash_key=True)
accountId = UnicodeAttribute()
configuration = MapAttribute()
Tags = MapAttribute()
version = HistoricalDecimalAttribute()
userIdentity = MapAttribute(null=True)
principalId = UnicodeAttribute(null=True)
userAgent = UnicodeAttribute(null=True)
sourceIpAddress = UnicodeAttribute(null=True)
requestParameters = MapAttribute(null=True)
class SignatureModel(BaseModel): # pylint: disable=too-many-instance-attributes
"""
Represents an signature in the database.
"""
class Meta:
"""Meta class for Signature."""
table_name = 'cla-{}-signatures'.format(stage)
if stage == 'local':
host = 'http://localhost:8000'
write_capacity_units = int(cla.conf['DYNAMO_WRITE_UNITS'])
read_capacity_units = int(cla.conf['DYNAMO_READ_UNITS'])
signature_id = UnicodeAttribute(hash_key=True)
signature_external_id = UnicodeAttribute(null=True)
signature_project_id = UnicodeAttribute()
signature_document_minor_version = NumberAttribute()
signature_document_major_version = NumberAttribute()
signature_reference_id = UnicodeAttribute()
signature_reference_type = UnicodeAttribute()
signature_type = UnicodeAttribute(default='cla')
signature_signed = BooleanAttribute(default=False)
signature_approved = BooleanAttribute(default=False)
signature_sign_url = UnicodeAttribute(null=True)
signature_return_url = UnicodeAttribute(null=True)
signature_callback_url = UnicodeAttribute(null=True)
signature_user_ccla_company_id = UnicodeAttribute(null=True)
signature_project_index = ProjectSignatureIndex()
signature_reference_index = ReferenceSignatureIndex()
# Callback type refers to either Gerrit or GitHub
signature_return_url_type = UnicodeAttribute(null=True)
# whitelists are only used by CCLAs
class Meta:
"""Meta class for User."""
table_name = "cla-{}-users".format(stage)
if stage == "local":
host = "http://localhost:8000"
write_capacity_units = int(cla.conf["DYNAMO_WRITE_UNITS"])
read_capacity_units = int(cla.conf["DYNAMO_READ_UNITS"])
user_id = UnicodeAttribute(hash_key=True)
# User Emails are specifically GitHub Emails
user_external_id = UnicodeAttribute(null=True)
user_emails = UnicodeSetAttribute(default=set())
user_name = UnicodeAttribute(null=True)
user_company_id = UnicodeAttribute(null=True)
user_github_id = NumberAttribute(null=True)
user_github_username = UnicodeAttribute(null=True)
user_github_username_index = GitHubUsernameIndex()
user_ldap_id = UnicodeAttribute(null=True)
user_github_id_index = GitHubUserIndex()
github_user_external_id_index = GithubUserExternalIndex()
note = UnicodeAttribute(null=True)
lf_email = UnicodeAttribute(null=True)
lf_username = UnicodeAttribute(null=True)
lf_username_index = LFUsernameIndex()
lf_sub = UnicodeAttribute(null=True)
class User(model_interfaces.User): # pylint: disable=too-many-public-methods
"""
ORM-agnostic wrapper for the DynamoDB User model.
"""
def set_document_tab_height(self, tab_height):
self.model.document_tab_height = tab_height
def set_document_tab_is_locked(self, is_locked):
self.model.document_tab_is_locked = is_locked
class DocumentModel(MapAttribute):
"""
Represents a document in the project model.
"""
document_name = UnicodeAttribute()
document_file_id = UnicodeAttribute(null=True)
document_content_type = UnicodeAttribute() # pdf, url+pdf, storage+pdf, etc
document_content = UnicodeAttribute(null=True) # None if using storage service.
document_major_version = NumberAttribute(default=1)
document_minor_version = NumberAttribute(default=0)
document_author_name = UnicodeAttribute()
# Not using UTCDateTimeAttribute due to https://github.com/pynamodb/PynamoDB/issues/162
document_creation_date = UnicodeAttribute()
document_preamble = UnicodeAttribute(null=True)
document_legal_entity_name = UnicodeAttribute(null=True)
document_tabs = ListAttribute(of=DocumentTabModel, default=[])
class Document(model_interfaces.Document):
"""
ORM-agnostic wrapper for the DynamoDB Document model.
"""
def __init__(self, # pylint: disable=too-many-arguments
document_name=None,
document_file_id=None,
document_content_type=None,
def __iter__(self):
"""Properly serialize the PynamoDB object as a `dict` via this function.
Helper for serializing into a typical `dict`. See: https://github.com/pynamodb/PynamoDB/issues/152
"""
for name, attr in self.get_attributes().items():
try:
if isinstance(attr, MapAttribute):
name, obj = name, getattr(self, name).as_dict()
yield name, fix_decimals(obj) # Don't forget to remove the stupid decimals :/
elif isinstance(attr, NumberAttribute) or isinstance(attr, HistoricalDecimalAttribute):
yield name, int(attr.serialize(getattr(self, name)))
elif isinstance(attr, ListAttribute):
name, obj = name, [el.as_dict() for el in getattr(self, name)]
yield name, fix_decimals(obj) # Don't forget to remove the stupid decimals :/
else:
yield name, attr.serialize(getattr(self, name))
# For Nulls:
except AttributeError:
yield name, None
class StoreModel(Model):
"""
Represents a key-value store in a DynamoDB.
"""
class Meta:
"""Meta class for Store."""
table_name = 'cla-{}-store'.format(stage)
if stage == 'local':
host = 'http://localhost:8000'
write_capacity_units = int(cla.conf['DYNAMO_WRITE_UNITS'])
read_capacity_units = int(cla.conf['DYNAMO_READ_UNITS'])
key = UnicodeAttribute(hash_key=True)
value = JSONAttribute()
expire = NumberAttribute()
class Store(key_value_store_interface.KeyValueStore):
"""
ORM-agnostic wrapper for the DynamoDB key-value store model.
"""
def __init__(self):
super(Store).__init__()
def set(self, key, value):
model = StoreModel()
model.key = key
model.value = value
model.expire = self.get_expire_timestamp()
model.save()
def get(self, key):
print(item)
print("-"*80)
# A model that uses aliased attribute names
class AliasedModel(Model):
class Meta:
table_name = "AliasedModel"
host = "http://localhost:8000"
forum_name = UnicodeAttribute(hash_key=True, attr_name='fn')
subject = UnicodeAttribute(range_key=True, attr_name='s')
views = NumberAttribute(default=0, attr_name='v')
replies = NumberAttribute(default=0, attr_name='rp')
answered = NumberAttribute(default=0, attr_name='an')
tags = UnicodeSetAttribute(attr_name='t')
last_post_datetime = UTCDateTimeAttribute(attr_name='lp')
if not AliasedModel.exists():
AliasedModel.create_table(read_capacity_units=1, write_capacity_units=1, wait=True)
# Create a thread
thread_item = AliasedModel(
'Some Forum',
'Some Subject',
tags=['foo', 'bar'],
last_post_datetime=datetime.now()
)
# Save the thread
thread_item.save()
def create_tables_if_not_exist():
if not Cajero.exists():
Cajero.create_table(read_capacity_units=5, write_capacity_units=2, wait=True)
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']