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
|
# coding: utf-8
from coreapi.codecs.base import BaseCodec
from coreapi.compat import urlparse
from coreapi.document import Document, Link, Field
from coreapi.exceptions import ParseError
from jsonhyperschema_codec.utils import _get_string, _get_list, _get_dict, get_dicts, _dereference
import json
import uritemplate
try:
from urllib.parse import unquote
except ImportError:
from urllib import unquote
__version__ = "1.0.3"
def _get_content(data, base_url, ref):
content = {}
links = _get_list(data, 'links')
properties = _get_dict(data, 'properties')
if properties:
for key, value in properties.items():
if not isinstance(value, dict):
continue
if list(value.keys()) == ['$ref']:
value = _dereference(value['$ref'], ref)
sub_content = _get_content(value, base_url, ref)
if sub_content:
content[key] = sub_content
if links:
for link in get_dicts(links):
rel = _get_string(link, 'rel')
if rel:
href = _get_string(link, 'href')
method = _get_string(link, 'method')
schema = _get_dict(link, 'schema')
schema_type = _get_list(schema, 'type')
schema_properties = _get_dict(schema, 'properties')
schema_required = _get_list(schema, 'required')
fields = []
url = urlparse.urljoin(base_url, href)
templated = uritemplate.variables(url)
for item in templated:
orig = item
if item.startswith('(') and item.endswith(')'):
item = unquote(item.strip('(').rstrip(')'))
if item.startswith('#/'):
components = [
component for component in item.strip('#/').split('/')
if component != 'definitions'
]
item = '_'.join(components).replace('-', '_')
url = url.replace(orig, item)
fields.append(Field(name=item, location='path', required=True))
if schema_type == ['object'] and schema_properties:
fields += [
Field(name=key, required=(key in schema_required))
for key in schema_properties.keys()
]
if rel == 'self':
rel = 'read'
content[rel] = Link(url=url, action=method, fields=fields)
return content
def _primative_to_document(data, base_url):
url = base_url
# Determine if the document contains a self URL.
links = _get_list(data, 'links')
for link in get_dicts(links):
href = _get_string(link, 'href')
rel = _get_string(link, 'rel')
if rel == 'self' and href:
url = urlparse.urljoin(url, href)
# Load the document content.
title = _get_string(data, 'title')
content = _get_content(data, url, ref=data)
return Document(title=title, url=url, content=content)
class JSONHyperSchemaCodec(BaseCodec):
"""
JSON Hyper-Schema.
"""
media_type = 'application/schema+json'
supports = ['decoding']
def load(self, bytes, **kwargs):
"""
Takes a bytestring and returns a document.
"""
base_url = kwargs.get('base_url', None)
try:
data = json.loads(bytes.decode('utf-8'))
except ValueError as exc:
raise ParseError('Malformed JSON. %s' % exc)
doc = _primative_to_document(data, base_url)
if not (isinstance(doc, Document)):
raise ParseError('Top level node must be a document.')
return doc
|
