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_resource_collection():
class TestCollection(ResourceCollection):
foo = types.StringType(required=True)
bar = types.BooleanType(default=True)
def create_resources(self):
self.res1 = Resource("res1", "foo", foo=self.foo)
tc = TestCollection(foo="foo!")
assert tc.foo == "foo!"
assert tc.res1.foo == "foo!"
assert tc.res1.id == "${res1.foo.id}"
with pytest.raises(SCHEMATICS_EXCEPTIONS):
TestCollection()
def test_returns_partial_data_with_conversion_errors():
class User(Model):
name = StringType(required=True)
age = IntType(required=True)
account_level = IntType()
with pytest.raises(DataError) as exception:
User({"name": "Jóhann", "age": "100 years", "account_level": "3"})
partial_data = exception.value.partial_data
assert partial_data == {
"name": u"Jóhann",
"account_level": 3,
}
# -*- coding: utf-8 -*-
from schematics.types import StringType
from schematics.types.compound import ListType, DictType
from base.models import BaseModel
class NewsModel(BaseModel):
title = StringType()
content = StringType()
author = StringType() # email of author
comments = ListType(DictType, compound_field=StringType)
MONGO_COLLECTION = 'news'
config.namespace = "prod"
config.stacks = [vpc]
print dump(config)
"""
namespace = StringType(required=True)
namespace_delimiter = StringType(serialize_when_none=False)
stacker_bucket = StringType(serialize_when_none=False)
stacker_bucket_region = StringType(serialize_when_none=False)
stacker_cache_dir = StringType(serialize_when_none=False)
sys_path = StringType(serialize_when_none=False)
package_sources = ModelType(PackageSources, serialize_when_none=False)
service_role = StringType(serialize_when_none=False)
pre_build = ListType(ModelType(Hook), serialize_when_none=False)
post_build = ListType(ModelType(Hook), serialize_when_none=False)
pre_destroy = ListType(ModelType(Hook), serialize_when_none=False)
post_destroy = ListType(ModelType(Hook), serialize_when_none=False)
tags = DictType(StringType, serialize_when_none=False)
# coding: utf-8
from schematics.models import Model
from schematics.types import DictType, StringType, ModelType
# dict of string
# dict of model
class A(Model):
s = StringType()
class B(Model):
sd = DictType(StringType)
md = DictType(ModelType(A))
def main():
b = B()
b.sd = {
"e": "ee",
"f": "ff"
}
b.md = {
"g": A({"s": "ss"})
}
def claim_models():
print("=== claim_models ===")
class Msg(Model):
type_ = StringType()
class TextMsg(Msg):
text = StringType()
class ImageMsg(Msg):
image = StringType()
def claim_func(field, data):
if data['type_'] == 'text':
return TextMsg
elif data['type_'] == 'image':
return ImageMsg
else:
return None
class MyModel(Model):
msg = PolyModelType(Msg, claim_function=claim_func)
m3 = MyModel({'msg': {'type_': 'text', 'text': 'ttt'}}) # schematics.exceptions.DataError: {"msg": {"image": "Rogue field"}}
raise ValidationError(u'This field is required.')
if relatedLot and isinstance(data['__parent__'], Model) and relatedLot not in [i.id for i in data['__parent__'].lots]:
raise ValidationError(u"relatedLot should be one of lots")
class Contract(Model):
class Options:
roles = {
'create': blacklist('id', 'status', 'documents', 'dateSigned'),
'edit': blacklist('id', 'documents', 'dateSigned'),
'embedded': schematics_embedded_role,
'view': schematics_default_role,
}
id = MD5Type(required=True, default=lambda: uuid4().hex)
awardID = StringType(required=True)
title = StringType() # Contract title
title_en = StringType()
title_ru = StringType()
description = StringType() # Contract description
description_en = StringType()
description_ru = StringType()
status = StringType(choices=['pending', 'terminated', 'active', 'cancelled'], default='pending')
period = ModelType(Period)
value = ModelType(Value)
dateSigned = IsoDateTimeType()
documents = ListType(ModelType(Document), default=list())
def validate_awardID(self, data, awardID):
if awardID and isinstance(data['__parent__'], Model) and awardID not in [i.id for i in data['__parent__'].awards]:
raise ValidationError(u"awardID should be one of awards")
sensitive_fields = ('__parent__', 'owner_token', 'transfer_token')
sensitive_embedded_role = SchematicsDocument.Options.roles['embedded'] + sensitive_fields
class Revision(Model):
author = StringType()
date = IsoDateTimeType(default=get_now)
changes = ListType(DictType(BaseType), default=list())
rev = StringType()
class BaseResourceItem(SchematicsDocument, Model):
owner = StringType() # the broker
owner_token = StringType() # token for broker access
transfer_token = SHA512Type() # token wich allows you to change the broker
mode = StringType(choices=['test']) # need for switching auction to different states
dateModified = IsoDateTimeType()
_attachments = DictType(DictType(BaseType), default=dict()) # couchdb attachments
revisions = ListType(ModelType(Revision), default=list()) # couchdb rev
__name__ = ''
def __repr__(self):
return '<%s:%r@%r>' % (type(self).__name__, self.id, self.rev)
@serializable(serialized_name='id')
def doc_id(self):
"""A property that is serialized by schematics exports."""
return self._id
model_id = IntType(serialized_name='modelId')
created = DateTimeType()
name = StringType(required=True)
source = StringType(required=True)
dockerhub_url = StringType(serialized_name='dockerhubUrl')
class PredictionDTO(Model):
""" Describes JSON of a set of predictions from a model """
prediction_id = IntType(serialized_name='predictionsId')
created = DateTimeType()
model_id = IntType(serialized_name='modelId', required=True)
version_id = IntType(serialized_name='versionId', required=True)
version_string = StringType(serialized_name='versionString')
dockerhub_hash = StringType(serialized_name='dockerhubHash')
bbox = ListType(FloatType, required=True)
tile_zoom = IntType(serialized_name='tileZoom', required=True)
class MLModelVersionDTO(Model):
""" Describes JSON of a ML model version """
version_id = IntType(serialized_name='versionId')
created = DateTimeType()
model_id = IntType(serialized_name='modelId', required=True)
version_major = IntType(serialized_name='versionMajor', required=True)
version_minor = IntType(serialized_name='versionMinor', required=True)
version_patch = IntType(serialized_name='versionPatch', required=True)
class StatementItem(BetfairModel):
ref_id = StringType()
item_date = DateTimeType()
amount = FloatType()
balance = FloatType()
item_class = EnumType(constants.ItemClass)
item_class_data = DictType(StringType)
legacy_data = ModelType(StatementLegacyData)
class AccountDetailsResponse(BetfairModel):
currency_code = StringType()
first_name = StringType()
last_name = StringType()
locale_code = StringType()
region = StringType()
timezone = StringType()
discount_rate = FloatType()
points_balance = IntType()
country_code = StringType()
class AccountStatementReport(BetfairModel):
account_statement = ListType(ModelType(StatementItem))
more_available = BooleanType()
class CurrencyRate(BetfairModel):
currency_code = StringType()
rate = FloatType()