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_list_type():
class SchemaExample(schemas.Model):
string_list = schemas.ListType(schemas.StringType)
m = SchemaExample()
assert m.import_data({"string_list": "string"}).to_primitive() == {"string_list": ["string"]}
assert m.import_data({"string_list": ["string1", "string2"]}).to_primitive() == {"string_list": ["string1", "string2"]}
class DataPoint(Model):
uid = StringType(required=True)
date = DateTimeType(required=True)
latitude = FloatType(min_value=-90, max_value=90, required=True)
longitude = FloatType(min_value=-180, max_value=180, required=True)
initiated = DateTimeType()
completed = DateTimeType()
# @todo: Support custom dimensions from signal
class SignalDataPoints(Model):
id = StringType(required=True)
data_points = ListType(ModelType(DataPoint), required=True)
chunk_size = IntType(default=1000, max_value=5000, required=True)
class DateDimension(Model):
type = StringType()
count = IntType()
min = DateTimeType()
max = DateTimeType()
class GeoDimension(Model):
type = StringType()
bbox = ListType(FloatType)
class Location(StringModel):
import_format = r'@(?P-?\d+(\.\d+)?),(?P-?\d+(\.\d+)?)'
export_format = "@{latitude},{longitude}"
latitude = FloatType(required=True)
longitude = FloatType(required=True)
class Place(Model):
class Options:
serialize_when_none = False
scope = ListType(StringType)
exact = ListType(StringType)
class Signal(Model):
class Options:
serialize_when_none = False
id = StringType(required=True)
analysis_from = DateTimeType()
analysis_to = DateTimeType()
analysis_tz = StringType(choices=pytz.all_timezones)
significance = FloatType(min_value=0, max_value=100)
metric = StringType(choices=('demand', 'lead', 'span'))
explain = DateType()
duration = IntType()
category = StringType()
labels = ListType(StringType())
country = StringType()
rank = IntType()
# `local_rank`, `aviation_rank`, and `phq_attendance` are paid features.
# They will only show up in your response body if you
# have subscribed to them.
local_rank = IntType()
aviation_rank = IntType()
phq_attendance = IntType()
entities = ListType(ModelType(Entities))
location = GeoJSONPointType()
place_hierarchies = ListType(ListType(StringType()))
scope = StringType()
relevance = FloatType()
state = StringType()
first_seen = DateTimeType()
updated = DateTimeType()
deleted_reason = StringType()
duplicate_of_id = StringType()
class EventResultSet(ResultSet):
overflow = BooleanType()
results = ResultType(Event)
from predicthq.endpoints.schemas import (
LimitMixin, Model, ResultSet, ListType, StringType, GeoJSONPointType, StringListType,
StringModelType, Location, DateTimeType, ResultType, SchematicsValidationError
)
class SearchParams(LimitMixin, Model):
class Options:
serialize_when_none = False
q = StringType()
id = ListType(StringType)
location = StringListType(StringModelType(Location), separator="+")
country = ListType(StringType)
type = ListType(StringType(choices=('planet', 'continent', 'country', 'region', 'county', 'local', 'major', 'metro', 'all')))
def validate(self, *args, **kwargs):
super(SearchParams, self).validate(*args, **kwargs)
if not any((self.q, self.id, self.location, self.country)):
raise SchematicsValidationError("Places search requires one of q, id, location or country")
class Place(Model):
id = StringType()
type = StringType()
name = StringType()
county = StringType()
def _coerce(self, value):
if not isinstance(value, (list, tuple)):
return [value]
else:
return super(ListType, self)._coerce(value)
rank_level = ListType(IntType(min_value=1, max_value=5))
# `local_rank`, `aviation_rank`, and `phq_attendance` are paid features.
# If you haven't subscribed to a paid feature, using it as a
# search param will have no effect on your search results.
local_rank = ModelType(IntRange)
local_rank_level = ListType(IntType(min_value=1, max_value=5))
aviation_rank = ModelType(IntRange)
aviation_rank_level = ListType(IntType(min_value=1, max_value=5))
phq_attendance = ModelType(IntRange)
country = ListType(StringType)
location_around = ModelType(LocationAround)
within = StringListType(StringModelType(Area), separator="+")
place = ModelType(Place)
relevance = ListType(StringType)
brand_unsafe = ModelType(BrandUnsafe)
entity = ModelType(Entity)
class Entities(Model):
class Options:
serialize_when_none = True
entity_id = StringType()
name = StringType()
type = StringType()
formatted_address = StringType()
class Event(Model):
end = ModelType(DateTimeRange)
end_around = ModelType(DateAround)
active = ModelType(DateTimeRange)
updated = ModelType(DateTimeRange)
state = StringType(choices=('active', 'deleted'))
deleted_reason = StringType(choices=('cancelled', 'duplicate', 'invalid', 'postponed'))
rank = ModelType(IntRange)
rank_level = ListType(IntType(min_value=1, max_value=5))
# `local_rank`, `aviation_rank`, and `phq_attendance` are paid features.
# If you haven't subscribed to a paid feature, using it as a
# search param will have no effect on your search results.
local_rank = ModelType(IntRange)
local_rank_level = ListType(IntType(min_value=1, max_value=5))
aviation_rank = ModelType(IntRange)
aviation_rank_level = ListType(IntType(min_value=1, max_value=5))
phq_attendance = ModelType(IntRange)
country = ListType(StringType)
location_around = ModelType(LocationAround)
within = StringListType(StringModelType(Area), separator="+")
place = ModelType(Place)
relevance = ListType(StringType)
brand_unsafe = ModelType(BrandUnsafe)
entity = ModelType(Entity)
class Entities(Model):
class Options:
serialize_when_none = True
PaginatedMixin, SortableMixin, Model, ResultSet, ListType, StringType, GeoJSONPointType,
StringListType, StringModelType, Area, ModelType, IntRange, IntType, DateTimeRange,
DateTimeType, FloatType, ResultType, DictType, DateType, Place, DateAround,
LocationAround, BooleanType, BrandUnsafe, Entity
)
class SearchParams(PaginatedMixin, SortableMixin, Model):
class Options:
serialize_when_none = False
id = ListType(StringType)
q = StringType()
label = ListType(StringType)
category = ListType(StringType)
start = ModelType(DateTimeRange)
start_around = ModelType(DateAround)
end = ModelType(DateTimeRange)
end_around = ModelType(DateAround)
active = ModelType(DateTimeRange)
updated = ModelType(DateTimeRange)
state = StringType(choices=('active', 'deleted'))
deleted_reason = StringType(choices=('cancelled', 'duplicate', 'invalid', 'postponed'))
rank = ModelType(IntRange)
rank_level = ListType(IntType(min_value=1, max_value=5))
# `local_rank`, `aviation_rank`, and `phq_attendance` are paid features.
# If you haven't subscribed to a paid feature, using it as a
# search param will have no effect on your search results.
local_rank = ModelType(IntRange)
local_rank_level = ListType(IntType(min_value=1, max_value=5))
limit = IntType(min_value=1)
class OffsetMixin(Model):
offset = IntType(min_value=0)
class PaginatedMixin(LimitMixin, OffsetMixin):
pass
class SortableMixin(Model):
sort = ListType(StringType())
class ResultSet(Model):
count = IntType()
previous = URLType()
next = URLType()
@property
def results(self): # pragma: nocover
raise NotImplementedError()
def _parse_params(self, url):
return dict(six.moves.urllib.parse.parse_qsl(six.moves.urllib.parse.urlparse(url).query))
def has_previous(self):