import importlib
import inspect
import random
import warnings
import six
from copy import copy
NoneType = type(None)
_ATTRIBUTE_CLASH_WARNING = (
"Property {attr!r} of Schema {schema} clashes with another "
"attribute of the same name in superclass {base}. This can lead to "
"surprising behavior when attempting to access that property on "
"an instance."
)
def flatten(xs):
"""Flatten a nested list or tuple.
Parameters:
xs (list|tuple): The input list or tuple.
Returns:
generator
"""
for x in xs:
if isinstance(x, (tuple, list)):
for y in flatten(x):
yield y
else:
yield x
class schema(type):
"""The metaclass for Schemas.
"""
def __new__(cls, classname, bases, attrs):
if classname in ("Schema",):
return super(schema, cls).__new__(cls, classname, bases, attrs)
properties = {}
for base in bases:
if issubclass(base, Schema) and base is not Schema:
properties = dict(list(base._properties.items()) + list(properties.items()))
erased_attrs = []
for name, value in list(attrs.items()):
if name in Schema.RESERVED_ATTRIBUTES:
raise AttributeError("'{}' is a reserved attribute on Schema".format(name))
if isinstance(value, SchemaLike):
properties[name] = value
if value._name is None:
value._name = name
del attrs[name]
erased_attrs.append(name)
for base in bases:
for attr in erased_attrs:
# If base has the attribute it means that either the
# base isn't a Schema or that the attribute was not
# erased (i.e. its type is not SchemaLike).
if hasattr(base, attr):
warnings.warn(_ATTRIBUTE_CLASH_WARNING.format(
attr=attr,
schema=classname,
base=base.__name__
), stacklevel=2)
attrs.update(_properties=properties)
return super(schema, cls).__new__(cls, classname, bases, attrs)
def __repr__(cls):
return "<@{} properties={!r}>".format(cls.__name__, cls._properties)
class propless_schema(schema):
"""The metaclass for Alternatives. This exists as a way to
differentiate between normal schema instances and Alternatives
during the adaptation process.
"""
def __new__(cls, classname, bases, attrs):
clazz = super(propless_schema, cls).__new__(cls, classname, bases, attrs)
assert not clazz._properties, \
"Alternative subclasses may not have properties of their own."
return clazz
class SchemaLike(object):
"""Base class for objects that describe the structure of data.
Parameters:
_name (str): This represents the object's serialized name. If
this is provided then this string will be used during
attribute lookups.
_description (str): This represents the object's description
when it gets adapted to some output format such as JSONSchema.
_optional (bool): When embedded as a property in a Schema this
attribute determines whether or not that property is
optional. This applies to any SchemaLike instance.
"""
_value = None
def __init__(self, _name=None, _description=None, _optional=None):
self._name = _name
self._description = _description
self._optional = _optional
self._metadata = ("_description",)
def __copy__(self):
clazz = type(self)
instance = clazz.__new__(clazz)
instance.__dict__.update({
name: copy(value) if isinstance(value, SchemaLike) else value
for name, value in six.iteritems(self.__dict__)
})
return instance
def lookup_name(self, default=None):
"""Look up this instance's virtual name, falling back to its
attribute name and then to ``default``.
Parameters:
default (str): The default value to return when this
instance doesn't define a virtual name AND when it's not a
part of a larger schema (where it would get assigned the
same name as its attribute name automatically).
Returns:
str: The name.
"""
return self._name or default
def validate(self, data): # pragma: no coverage
"""Validate ``data`` against this instance.
Parameters:
value (object): The object to validate.
Returns:
None: On success.
"""
raise NotImplementedError
def generate(self): # pragma: no coverage
"""Generate a random value that satisifies this SchemaLike.
Note:
Certain SchemaLikes may break this invariant if it isn't
feasible to automatically generate valid data.
"""
raise NotImplementedError
def update(self, value): # pragma: no coverage
"""Update this instance's value.
Note:
If you want to both validate and update a SchemaLike based
on some data then you should use this method directly as it
validates the input data before updating the instance.
Raises:
ValidationError
Parameters:
value (object): The object that will replace this instance's
underlying value. This value will be validated prior to
assignment.
Returns:
None: On success.
"""
raise NotImplementedError
def virtualize(self):
"""Return a virtual value for easy access to the underlying
data. Most :class:`.SchemaLike` instances simply return their
materialized value.
"""
return self.materialize()
def materialize(self, sparse=False): # pragma: no coverage
"""Return the concrete value for this instance.
"""
raise NotImplementedError
def __eq__(self, other):
return isinstance(other, self.__class__) and self.materialize() == other.materialize()
def __ne__(self, other):
return not (self == other)
class Schema(six.with_metaclass(schema, SchemaLike)):
"""Schema describe the structure of data. Schema instances can be
treated like models in an ORM.
Parameters:
_name (str): This represents the object's serialized name. If
this is provided then this string will be used during
attribute lookups.
_description (str): This represents the object's description
when it gets adapted to some output format such as JSONSchema.
_optional (bool): When embedded as a property in a Schema this
attribute determines whether or not that property is
optional. This applies to any SchemaLike instance.
Examples:
>>> from schema import *
>>> class Meta(Schema):
... id = String()
>>> class Resource(Schema):
... _meta = Meta()
>>> class User(Resource):
... name = String()
>>> m = Meta(id="a-user")
<#Meta properties={'id': <.String value='a-user'>}>
>>> u = User(_meta=m, name="A User")
<#User properties={'_meta': <#Meta properties={'id': <.String value='a-user'}, 'name': <.String value='A User'}>
>>> m.materialize()
{'id': 'a-user'}
>>> u.materialize()
{'_meta': {'id': 'a-user"}, 'name': 'A User'}
>>> u.validate({})
Traceback (most recent call last):
...
ValidationError: ['_meta: missing', 'name: missing']
>>> u.validate({"_meta": {"id": 42}})
Traceback (most recent call last):
...
ValidationError: ['_meta.id: invalid string', 'name: missing']
>>> u.validate({"_meta": {"id": "a-user"}, "name": "A user"})
>>> u.meta.id
'a-user'
>>> u.name
'A user'
"""
RESERVED_ATTRIBUTES = (
"_optional",
"_description",
"_metadata",
"_properties",
"_contains_data",
)
"""The list of attributes that a materialized Schema may not
contain. This exists in order to protect users from property
access bugs on :class:`.Schema` instances.
"""
def __init__(self, _optional=False, _description=None, **properties):
super(Schema, self).__init__(
_optional=_optional,
_description=_description
)
self._initialize()
for name, value in six.iteritems(properties):
setattr(self, name, value)
def __copy__(self):
instance = super(Schema, self).__copy__()
instance._initialize_properties()
return instance
def _initialize(self):
self._contains_data = False
self._initialize_properties()
def _initialize_properties(self):
self._properties = {
name: copy(prop) for name, prop in six.iteritems(self.__class__._properties)
}
def validate(self, data):
"""Validate ``data`` against this :class:`.Schema`.
Raises:
ValidationError: When the dictionary fails to validate.
Parameters:
data (dict) A dictionary of data to validate.
Returns:
None: On success.
"""
if data is None:
if self._optional:
return
raise ValidationError("null value encountered")
if not isinstance(data, dict):
raise ValidationError("invalid input {!r}".format(data))
errors = []
for key, validation in six.iteritems(self._properties):
key = validation.lookup_name(key)
try:
value = data.get(key, None)
validation.validate(value)
except ValidationError as e:
if isinstance(e.message, (tuple, list)):
messages = flatten(e.message)
errors.extend("{}.{}".format(key, m) for m in messages)
else:
errors.append("{}: {}".format(key, e.message))
if errors:
raise ValidationError(errors)
def generate(self):
name = self.__class__.__name__
data = {p.lookup_name(k): p.generate() for k, p in six.iteritems(self._properties)}
if "kind" in self._properties and name.lower() in self._properties:
union = self._properties[name.lower()]
for v in union.validations:
if v.__class__.__name__ == data["kind"]:
data[name.lower()] = v.generate()
break
return data
def virtualize(self):
"""The virtual representation of a Schema is that Schema.
"""
return self
def materialize(self, sparse=False):
"""Materializes this :class:`.Schema` into a dictionary
recursively.
Parameters:
sparse (bool): Controls whether or not null properties
should be included in the result.
Returns:
dict: The dictionary representation of this instance.
"""
if self._optional and not self._contains_data:
return None
data = {}
for k, p in six.iteritems(self._properties):
value = p.materialize(sparse=sparse)
if not sparse or value is not None:
data[p.lookup_name(k)] = value
return data
def update(self, data):
"""Validate ``data`` against this :class:`.Schema` and set its
properties to the appropriate values from the data dictionary
on succcess.
Raises:
ValidationError: If `data` fails to :meth:`.Schema.validate`.
Parameters:
data(dict):
Returns:
None: On success.
"""
assert isinstance(data, (dict, NoneType))
if data is None:
if not self._optional:
raise ValidationError("null value encountered")
return self._initialize()
# This is used to keep track of whether or not a Schema should
# materialize to None.
self._contains_data = bool(data)
errors = []
for name, prop in six.iteritems(self._properties):
extern_name = prop.lookup_name(name)
try:
setattr(self, name, data.get(extern_name))
except ValidationError as e:
if isinstance(e.message, (tuple, list)):
messages = flatten(e.message)
errors.extend(messages)
else:
errors.append(e.message)
if errors:
raise ValidationError(errors)
def fetch_property(self, name):
property = self._properties.get(name, None)
if property is None:
raise AttributeError("no attribute called '{}'".format(name))
return property
def __setattr__(self, name, value):
if name in self._properties:
property = self.fetch_property(name)
try:
if isinstance(value, (Schema, Property)):
property.update(value.materialize())
else:
property.update(value)
return
except ValidationError as e:
extern_name = property.lookup_name(name)
if isinstance(e.message, (tuple, list)):
messages = flatten(e.message)
raise ValidationError(["{}.{}".format(extern_name, m) for m in messages])
else:
raise ValidationError("{}: {}".format(extern_name, e.message))
return super(Schema, self).__setattr__(name, value)
def __getattr__(self, name):
return self.fetch_property(name).virtualize()
def __repr__(self):
return "<#{} properties={!r}>".format(self.__class__.__name__, self._properties)
def Alternative(*schemas):
"""Unifies a list of :class:`.Schema` subclasses into one, selecting
the most appropriate one on each call to :meth:`.SchemaLike.validate`.
See the examples for more information.
Note:
You may not specify additional properties on an Alternative and
it is your responsibility to ensure that the schemas you provide
it with upon creation are not ambiguous when validated against
each other's data.
Examples:
>>> from schema import *
>>> class ResetPasswordRequest(Schema):
... password = String()
... reset_token = String(_name="token")
>>> class UpdateUsernameRequest(Schema):
... username = String()
>>> class PatchUserRequest(Alternative(
... UpdateUsernameRequest,
... ResetPasswordRequest
... )):
... pass
>>> r = PatchUserRequest()
>>> request = r.validate({"username": "new-username"})
>>> r.validate({"password": "a-password", "token": "foo"})
ResetPasswordRequest(...)
>>> r.validate({"username": "new-username"})
UpdateUsernameRequest(...)
>>> r.update({"username": "new-username"})
>>> r.username
"new-username"
>>> r.materialize()
{"username": "new-username"}
>>> isinstance(r._schema, UpdateUsernameRequest)
True
>>> isinstance(r._schema, ResetPasswordRequest)
False
Parameters:
schemas(Schema list): A list of :class:`.Schema` classes.
Returns:
Schema: A :class:`.Schema` representing the disjoint union of
the input Schema.
"""
for schema in schemas:
if not (inspect.isclass(schema) and issubclass(schema, Schema)):
raise TypeError("Alternative expects a list of Schema subclasses.")
class Alternative(six.with_metaclass(propless_schema, Schema)):
_alternatives = schemas
_schema = None
def validate(self, data):
errors = []
for schema in schemas:
try:
schema = schema()
schema.validate(data)
return schema
except ValidationError as e:
errors.append(e.message)
raise ValidationError(errors)
def generate(self):
return random.choice(schemas)().generate()
def update(self, data):
self._schema = self.validate(data)
self._properties = self._schema._properties
super(Alternative, self).update(data)
def __getattr__(self, name):
if self._schema is not None:
return getattr(self._schema, name)
raise AttributeError("no attribute called {!r}", name)
return Alternative
class LazyLike(object):
"""Base class for objects that act like Lazy schemas.
"""
def Lazy(schema):
"""Lazily refrence a schema using either its fully qualified name or
its name within the module ``Lazy`` is called in. Doing either of
these things will ensure that the schema is instantiated lazily.
Note:
Because of the way these schemas are lazy loaded you cannot
reference dynamically-generated schemas.
Note:
You must ensure you don't shadow the definition of ``schema``
within your module.
Note:
Un-qualified names rely on the Python implementation you're
running req on to have stack frame support.
Examples:
>>> class Nil(Schema):
... def validate(self, data):
... if data:
... raise ValidationError("Nil must be empty")
>>> class Cons(Schema):
... car = Integer()
... cdr = Or(Lazy("Cons"), Nil())
>>> class ConsQualified(Schema):
... car = Integer()
... cdr = Or(Lazy("tests.test_lazy.ConsQualified"), Nil())
>>> l = Cons()
>>> l.validate({"car": 1, "cdr": {"car": 2, "cdr": {}}})
>>> l.update({"car": 1, "cdr": {"car": 2, "cdr": {}}})
>>> l.validate({"car": 1, "cdr": {"car": "2", "cdr": {}}})
Traceback (most recent call last):
...
ValidationError: ['cdr.car: invalid integer for Cons schema']
Raises:
RuntimeError: When the object referenced by ``schema`` cannot be
found at runtime.
TypeError: When the object referenced by ``schema`` is not a
subclass of :class:`.SchemaLike`. This error is raised at
runtime.
Parameters:
schema(str): The name of a :class:`.Schema`.
Returns:
Schema: A proxy that will lazily instantiate the referenced
schema on its first use.
"""
if "." not in schema:
frame = inspect.stack()[1]
module = inspect.getmodule(frame[0])
del frame
else:
segments = schema.split(".")
module_path, schema = ".".join(segments[:-1]), segments[-1]
module = importlib.import_module(module_path)
class Lazy(LazyLike):
_instance = None
@property
def instance(self):
if self._instance is None:
clazz = getattr(module, schema, None)
if clazz is None:
raise RuntimeError(
"Lazy schema {!r} not found.".format(schema)
)
if not issubclass(clazz, SchemaLike):
raise TypeError(
"Lazy schema {!r} is not a subclass of SchemaLike. "
"Did you accidentally shadow its definition?".format(schema)
)
self._instance = clazz()
return self._instance
@property
def __class__(self):
return self.instance.__class__
def __deepcopy__(self, memo):
return Lazy()
def __copy__(self):
return copy(self.instance)
def __getattr__(self, key):
return getattr(self.instance, key)
return Lazy()
class Property(SchemaLike):
"""Base class for Properties.
Parameters:
_name (str): This represents the object's serialized name. If
this is provided then this string will be used during
attribute lookups.
_description (str): This represents the object's description
when it gets adapted to some output format such as JSONSchema.
_optional (bool): When embedded as a property in a Schema this
attribute determines whether or not that property is
optional. This applies to any SchemaLike instance.
"""
def _validate(self, data): # pragma: no coverage
"""All Property subclasses must implement this method.
"""
raise NotImplementedError
def validate(self, data):
if data is None:
if self._optional:
return
raise ValidationError("missing")
return self._validate(data)
def generate(self):
if self._optional and random.randint(1, 5) == 1:
return None
return self._generate()
def materialize(self, sparse=False):
return self._value
def update(self, value):
self.validate(value)
self._value = value
def __repr__(self):
return "<.{} value={!r}>".format(self.__class__.__name__, self.materialize())
class ValidationError(Exception):
"""Raised by :class:`SchemaLikes <.SchemaLike>` whenever some data
fails to validate.
Attributes:
message(str|str list): One or more error messages.
"""
def __init__(self, message):
self.message = message
super(ValidationError, self).__init__(message)
def __str__(self): # pragma: no cover
return str(self.message)