/
schema.py
121 lines (98 loc) · 3.98 KB
/
schema.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
import os
import copy
import six
import requests
import json
import jsonschema
import datapackage_registry
from datapackage_registry.exceptions import DataPackageRegistryException
from .exceptions import (
SchemaError, ValidationError
)
class Schema(object):
'''Abstracts a JSON Schema and allows validation of data against it.
Args:
schema (str or dict): The JSON Schema itself as a dict, or a local path
or URL to it.
Raises:
SchemaError: If unable to load schema or it was invalid.
Warning:
The schema objects created with this class are read-only. You should
change any of its attributes after creation.
'''
def __init__(self, schema):
self._registry = self._load_registry()
self._schema = self._load_schema(schema, self._registry)
self._validator = self._load_validator(self._schema, self._registry)
self._check_schema()
def to_dict(self):
'''dict: Convert this :class:`.Schema` to dict.'''
return copy.deepcopy(self._schema)
def validate(self, data):
'''Validates a data dict against this schema.
Args:
data (dict): The data to be validated.
Raises:
ValidationError: If the data is invalid.
'''
try:
self._validator.validate(data)
except jsonschema.ValidationError as e:
six.raise_from(ValidationError.create_from(e), e)
def _load_registry(self):
try:
return datapackage_registry.Registry()
except DataPackageRegistryException as e:
six.raise_from(SchemaError(e), e)
def _load_schema(self, schema, registry):
the_schema = schema
if isinstance(schema, six.string_types):
try:
the_schema = registry.get(schema)
if not the_schema:
if os.path.isfile(schema):
with open(schema, 'r') as f:
the_schema = json.load(f)
else:
req = requests.get(schema)
req.raise_for_status()
the_schema = req.json()
except (IOError,
ValueError,
DataPackageRegistryException,
requests.exceptions.RequestException) as e:
msg = 'Unable to load schema at "{0}"'
six.raise_from(SchemaError(msg.format(schema)), e)
elif isinstance(the_schema, dict):
the_schema = copy.deepcopy(the_schema)
else:
msg = 'Schema must be a "dict", but was a "{0}"'
raise SchemaError(msg.format(type(the_schema).__name__))
return the_schema
def _load_validator(self, schema, registry):
resolver = None
if registry.base_path:
path = 'file://{base_path}/'.format(base_path=registry.base_path)
resolver = jsonschema.RefResolver(path, schema)
validator_class = jsonschema.validators.validator_for(schema)
return validator_class(schema, resolver=resolver)
def _check_schema(self):
try:
self._validator.check_schema(self._schema)
except jsonschema.exceptions.SchemaError as e:
six.raise_from(SchemaError.create_from(e), e)
def __getattr__(self, name):
if name in self.__dict__.get('_schema', {}):
return copy.deepcopy(self._schema[name])
msg = '\'{0}\' object has no attribute \'{1}\''
raise AttributeError(msg.format(self.__class__.__name__, name))
def __setattr__(self, name, value):
if name in self.__dict__.get('_schema', {}):
raise AttributeError('can\'t set attribute')
super(self.__class__, self).__setattr__(name, value)
def __dir__(self):
return list(self.__dict__.keys()) + list(self._schema.keys())