forked from singer-io/singer-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathschema.py
More file actions
100 lines (84 loc) · 3.04 KB
/
schema.py
File metadata and controls
100 lines (84 loc) · 3.04 KB
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
# pylint: disable=redefined-builtin, too-many-arguments, invalid-name
'''Provides an object model for JSON Schema'''
import json
# These are standard keys defined in the JSON Schema spec
STANDARD_KEYS = [
'sqlDatatype',
'selected',
'inclusion',
'description',
'minimum',
'maximum',
'exclusiveMinimum',
'exclusiveMaximum',
'multipleOf',
'maxLength',
'format',
'type'
]
class Schema(object): # pylint: disable=too-many-instance-attributes
'''Object model for JSON Schema.
Tap and Target authors may find this to be more convenient than
working directly with JSON Schema data structures.
'''
def __init__(self, type=None, format=None, properties=None,
items=None, sqlDatatype=None, selected=None,
inclusion=None, description=None, minimum=None,
maximum=None, exclusiveMinimum=None,
exclusiveMaximum=None, multipleOf=None, maxLength=None):
self.type = type
self.properties = properties
self.items = items
self.sqlDatatype = sqlDatatype
self.selected = selected
self.inclusion = inclusion
self.description = description
self.minimum = minimum
self.maximum = maximum
self.exclusiveMinimum = exclusiveMinimum
self.exclusiveMaximum = exclusiveMaximum
self.multipleOf = multipleOf
self.maxLength = maxLength
self.format = format
def __str__(self):
return json.dumps(self.to_dict())
def __repr__(self):
pairs = [k + '=' + repr(v) for k, v in self.__dict__.items()]
args = ', '.join(pairs)
return 'Schema(' + args + ')'
def __eq__(self, other):
return self.__dict__ == other.__dict__
def to_dict(self):
'''Return the raw JSON Schema as a (possibly nested) dict.'''
result = {}
if self.properties:
result['properties'] = {
k: v.to_dict()
for k, v
in self.properties.items() # pylint: disable=no-member
}
if self.items:
result['items'] = self.items.to_dict() # pylint: disable=no-member
for key in STANDARD_KEYS:
if self.__dict__[key] is not None:
result[key] = self.__dict__[key]
return result
@classmethod
def from_dict(cls, data, **schema_defaults):
'''Initialize a Schema object based on the JSON Schema structure.
:param schema_defaults: The default values to the Schema
constructor.'''
kwargs = schema_defaults.copy()
properties = data.get('properties')
items = data.get('items')
if properties:
kwargs['properties'] = {
k: Schema.from_dict(v, **schema_defaults)
for k, v in properties.items()
}
if items:
kwargs['items'] = Schema.from_dict(items, **schema_defaults)
for key in STANDARD_KEYS:
if key in data:
kwargs[key] = data[key]
return Schema(**kwargs)