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_attribute_paths_wrapping(self):
class InnerMapAttribute(MapAttribute):
map_attr = MapAttribute(attr_name='dyn_map_attr')
class MiddleMapAttributeA(MapAttribute):
inner_map = InnerMapAttribute(attr_name='dyn_in_map_a')
class MiddleMapAttributeB(MapAttribute):
inner_map = InnerMapAttribute(attr_name='dyn_in_map_b')
class OuterMapAttribute(MapAttribute):
mid_map_a = MiddleMapAttributeA()
mid_map_b = MiddleMapAttributeB()
class MyModel(Model):
outer_map = OuterMapAttribute(attr_name='dyn_out_map')
mid_map_a_map_attr = MyModel.outer_map.mid_map_a.inner_map.map_attr
mid_map_b_map_attr = MyModel.outer_map.mid_map_b.inner_map.map_attr
assert mid_map_a_map_attr.attr_name == 'dyn_map_attr'
assert mid_map_a_map_attr.attr_path == ['dyn_out_map', 'mid_map_a', 'dyn_in_map_a', 'dyn_map_attr']
assert mid_map_b_map_attr.attr_name == 'dyn_map_attr'
assert mid_map_b_map_attr.attr_path == ['dyn_out_map', 'mid_map_b', 'dyn_in_map_b', 'dyn_map_attr']
car_id = NumberAttribute(null=False)
car_color = UnicodeAttribute(null=True)
car_info = CarInfoMap(null=True)
class OfficeEmployeeMap(MapAttribute):
office_employee_id = NumberAttribute()
person = Person()
office_location = Location()
def cool_function(self):
return 1
class GroceryList(Model):
class Meta:
table_name = 'GroceryListModel'
store_name = UnicodeAttribute(hash_key=True)
groceries = ListAttribute()
class Office(Model):
class Meta:
table_name = 'OfficeModel'
office_id = NumberAttribute(hash_key=True)
address = Location()
employees = ListAttribute(of=OfficeEmployeeMap)
class BooleanModel(Model):
user_id = NumberAttribute(hash_key=True)
created_at = UTCDateTimeAttribute(range_key=True, default=datetime.now())
amount = NumberAttribute()
currency = UnicodeAttribute()
class DifferentRegion(Model):
class Meta:
region = 'us-east-2'
table_name = 'different-region'
entry_index = NumberAttribute(hash_key=True)
class Foo(Model):
class Meta:
region = 'us-east-1'
table_name = 'foo'
bar = NumberAttribute(hash_key=True)
star = UnicodeAttribute(null=True)
version = VersionAttribute()
TEST_MODELS = [
BankStatement,
DifferentRegion,
LineItem,
User,
Foo
]
views = NumberAttribute(null=True)
is_active = BooleanAttribute(null=True)
signature = UnicodeAttribute(null=True)
ttl = TTLAttribute(null=True)
class CustomAttrIndex(LocalSecondaryIndex):
class Meta:
read_capacity_units = 2
write_capacity_units = 1
projection = AllProjection()
overidden_uid = UnicodeAttribute(hash_key=True, attr_name='user_id')
class CustomAttrNameModel(Model):
"""
A testing model
"""
class Meta:
table_name = 'CustomAttrModel'
overidden_user_name = UnicodeAttribute(hash_key=True, attr_name='user_name')
overidden_user_id = UnicodeAttribute(range_key=True, attr_name='user_id')
overidden_attr = UnicodeAttribute(attr_name='foo_attr', null=True)
uid_index = CustomAttrIndex()
class UserModel(Model):
"""
A testing model
class MapAttrSubClassWithRawMapAttr(MapAttribute):
num_field = NumberAttribute()
str_field = UnicodeAttribute()
map_field = MapAttribute()
class ExplicitRawMapAsMemberOfSubClass(Model):
class Meta:
table_name = 'ExplicitRawMapAsMemberOfSubClass'
map_id = NumberAttribute(hash_key=True)
sub_attr = MapAttrSubClassWithRawMapAttr()
class Animal(Model):
name = UnicodeAttribute(hash_key=True)
class Dog(Animal):
class Meta:
table_name = 'Dog'
breed = UnicodeAttribute()
class TTLModel(Model):
class Meta:
table_name = 'TTLModel'
user_name = UnicodeAttribute(hash_key=True)
my_ttl = TTLAttribute(default_for_new=timedelta(minutes=1))
@classmethod
def _conditional_operator_check(cls, conditional_operator):
pass
@property
def aggregate_id(self):
return self.aggregate_key.split(AGGREGATE_KEY_DELIMITER)[0]
@property
def version(self):
return int(self.aggregate_key.split(
AGGREGATE_KEY_DELIMITER
)[1])
class Snapshot(Model):
class Meta:
table_name = "snapshots"
host = os.getenv('DYNAMODB_HOST', None)
aggregate_name = UnicodeAttribute(hash_key=True)
aggregate_key = UnicodeAttribute(range_key=True)
created_at = UTCDateTimeAttribute()
state = MapAttribute()
@classmethod
def from_aggregate(cls, aggregate):
state = aggregate.current_state
last_event = aggregate.repository.get_last_event()
snapshot = cls(
aggregate_name=aggregate.aggregate_name,
aggregate_key=last_event.aggregate_id,
# Check if db is a Flywheel instance
if self.db_adapter is None: # pragma: no cover
try:
from flask_flywheel import Flywheel
if isinstance(db, Flywheel):
self.db_adapter = DynamoDbAdapter(app, db)
except ImportError:
pass # Ignore ImportErrors
# Check if the UserClass is a Pynamo Model
if self.db_adapter is None:
try:
from pynamodb.models import Model
if issubclass(UserClass, Model):
self.db_adapter = PynamoDbAdapter(app)
except ImportError:
pass # Ignore ImportErrors
# Check self.db_adapter
if self.db_adapter is None:
raise ConfigError(
'No Flask-SQLAlchemy, Flask-MongoEngine or Flask-Flywheel installed and no Pynamo Model in use.'\
' You must install one of these Flask extensions.')
class ReferenceSignatureIndex(GlobalSecondaryIndex):
"""
This class represents a global secondary index for querying signatures by reference.
"""
class Meta:
"""Meta class for reference Signature index."""
index_name = 'reference-signature-index'
write_capacity_units = int(cla.conf['DYNAMO_WRITE_UNITS'])
read_capacity_units = int(cla.conf['DYNAMO_READ_UNITS'])
# All attributes are projected - not sure if this is necessary.
projection = AllProjection()
# This attribute is the hash key for the index.
signature_reference_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:
companies_dict = [company_model.to_dict() for company_model in companies]
return companies_dict
def get_managers_by_company_acl(self, company_acl):
managers = []
user_model = User()
for username in company_acl:
users = user_model.get_user_by_username(str(username))
if len(users) > 1:
cla.log.warning(f"More than one user record returned for username: {username}")
if users is not None:
managers.append(users[0])
return managers
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()
"""
This class will serializer/deserialize any picklable Python object.
The value will be stored as a binary attribute in DynamoDB.
"""
def serialize(self, value):
"""
The super class takes the binary string returned from pickle.dumps
and encodes it for storage in DynamoDB
"""
return super(PickleAttribute, self).serialize(pickle.dumps(value))
def deserialize(self, value):
return pickle.loads(super(PickleAttribute, self).deserialize(value))
class CustomAttributeModel(Model):
"""
A model with a custom attribute
"""
class Meta:
host = 'http://localhost:8000'
table_name = 'custom_attr'
read_capacity_units = 1
write_capacity_units = 1
id = UnicodeAttribute(hash_key=True)
obj = PickleAttribute()
# Create the example table
if not CustomAttributeModel.exists():
CustomAttributeModel.create_table(wait=True)