-
Notifications
You must be signed in to change notification settings - Fork 46
/
api.py
305 lines (238 loc) · 10 KB
/
api.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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
# -*- coding: utf-8 -*-
"""
Module for automatically serving /api-docs* via Pyramid.
"""
from __future__ import absolute_import
import copy
import os.path
import simplejson
import yaml
from bravado_core.spec import strip_xscope
from six.moves.urllib.parse import urlparse
from six.moves.urllib.parse import urlunparse
from six.moves.urllib.request import pathname2url
from pyramid_swagger.model import PyramidEndpoint
# TODO: document that this is now a public interface
def register_api_doc_endpoints(config, endpoints, base_path='/api-docs'):
"""Create and register pyramid endpoints to service swagger api docs.
Routes and views will be registered on the `config` at `path`.
:param config: a pyramid configuration to register the new views and routes
:type config: :class:`pyramid.config.Configurator`
:param endpoints: a list of endpoints to register as routes and views
:type endpoints: a list of :class:`pyramid_swagger.model.PyramidEndpoint`
:param base_path: the base path used to register api doc endpoints.
Defaults to `/api-docs`.
:type base_path: string
"""
for endpoint in endpoints:
path = base_path.rstrip('/') + endpoint.path
config.add_route(endpoint.route_name, path)
config.add_view(
endpoint.view,
route_name=endpoint.route_name,
renderer=endpoint.renderer)
def build_swagger_12_endpoints(resource_listing, api_declarations):
"""
:param resource_listing: JSON representing a Swagger 1.2 resource listing
:type resource_listing: dict
:param api_declarations: JSON representing Swagger 1.2 api declarations
:type api_declarations: dict
:rtype: iterable of :class:`pyramid_swagger.model.PyramidEndpoint`
"""
yield build_swagger_12_resource_listing(resource_listing)
for name, filepath in api_declarations.items():
with open(filepath) as input_file:
yield build_swagger_12_api_declaration(
name, simplejson.load(input_file))
def build_swagger_12_resource_listing(resource_listing):
"""
:param resource_listing: JSON representing a Swagger 1.2 resource listing
:type resource_listing: dict
:rtype: :class:`pyramid_swagger.model.PyramidEndpoint`
"""
def view_for_resource_listing(request):
# Thanks to the magic of closures, this means we gracefully return JSON
# without file IO at request time.
return resource_listing
return PyramidEndpoint(
path='',
route_name='pyramid_swagger.swagger12.api_docs',
view=view_for_resource_listing,
renderer='json')
def build_swagger_12_api_declaration(resource_name, api_declaration):
"""
:param resource_name: The `path` parameter from the resource listing for
this resource.
:type resource_name: string
:param api_declaration: JSON representing a Swagger 1.2 api declaration
:type api_declaration: dict
:rtype: :class:`pyramid_swagger.model.PyramidEndpoint`
"""
# NOTE: This means our resource paths are currently constrained to be valid
# pyramid routes! (minus the leading /)
route_name = 'pyramid_swagger.swagger12.apidocs-{0}'.format(resource_name)
return PyramidEndpoint(
path='/{0}'.format(resource_name),
route_name=route_name,
view=build_swagger_12_api_declaration_view(api_declaration),
renderer='json')
def build_swagger_12_api_declaration_view(api_declaration_json):
"""Thanks to the magic of closures, this means we gracefully return JSON
without file IO at request time.
"""
def view_for_api_declaration(request):
# Note that we rewrite basePath to always point at this server's root.
return dict(
api_declaration_json,
basePath=str(request.application_url),
)
return view_for_api_declaration
class NodeWalker(object):
def __init__(self):
pass
def walk(self, item, *args, **kwargs):
dupe = copy.deepcopy(item)
return self._walk(dupe, *args, **kwargs)
def _walk(self, item, *args, **kwargs):
if isinstance(item, list):
return self._walk_list(item, *args, **kwargs)
elif isinstance(item, dict):
return self._walk_dict(item, *args, **kwargs)
else:
return self._walk_item(item, *args, **kwargs)
def _walk_list(self, item, *args, **kwargs):
for index, subitem in enumerate(item):
item[index] = self._walk(subitem, *args, **kwargs)
return item
def _walk_dict(self, item, *args, **kwargs):
for key, value in item.items():
item[key] = self._walk_dict_item(key, value, *args, **kwargs)
return item
def _walk_dict_item(self, key, value, *args, **kwargs):
return self._walk(value, *args, **kwargs)
def _walk_item(self, value, *args, **kwargs):
return value
def get_path_if_relative(url):
parts = urlparse(url)
if parts.scheme or parts.netloc:
# only rewrite relative paths
return
if not parts.path:
# don't rewrite internal refs
return
if parts.path.startswith('/'):
# don't rewrite absolute refs
return
return parts
class NodeWalkerForRefFiles(NodeWalker):
def walk(self, spec):
all_refs = []
spec_fname = spec.origin_url
if spec_fname.startswith('file://'):
spec_fname = spec_fname.replace('file://', '')
spec_dirname = os.path.dirname(spec_fname)
parent = super(NodeWalkerForRefFiles, self)
parent.walk(spec.client_spec_dict, spec, spec_dirname, all_refs)
all_refs = [os.path.relpath(f, spec_dirname) for f in all_refs]
all_refs = set(all_refs)
core_dirname, core_fname = os.path.split(spec_fname)
all_refs.add(core_fname)
return all_refs
def _walk_dict_item(self, key, value, spec, dirname, all_refs):
if key != '$ref':
parent = super(NodeWalkerForRefFiles, self)
return parent._walk_dict_item(key, value, spec, dirname, all_refs)
# assume $ref is the only key in the dict
parts = get_path_if_relative(value)
if not parts:
return value
full_fname = os.path.join(dirname, parts.path)
norm_fname = os.path.normpath(full_fname)
all_refs.append(norm_fname)
with spec.resolver.resolving(value) as spec_dict:
dupe = copy.deepcopy(spec_dict)
self._walk(dupe, spec, os.path.dirname(norm_fname), all_refs)
class NodeWalkerForCleaningRefs(NodeWalker):
def walk(self, item, schema_format):
parent = super(NodeWalkerForCleaningRefs, self)
return parent.walk(item, schema_format)
@staticmethod
def fix_ref(ref, schema_format):
parts = get_path_if_relative(ref)
if not parts:
return
path, ext = os.path.splitext(parts.path)
return urlunparse([
parts.scheme,
parts.netloc,
'{0}.{1}'.format(path, schema_format),
parts.params,
parts.query,
parts.fragment,
])
def _walk_dict_item(self, key, value, schema_format):
if key != '$ref':
parent = super(NodeWalkerForCleaningRefs, self)
return parent._walk_dict_item(key, value, schema_format)
return self.fix_ref(value, schema_format) or value
class YamlRendererFactory(object):
def __init__(self, info):
pass
def __call__(self, value, system):
response = system['request'].response
response.headers['Content-Type'] = 'application/x-yaml; charset=UTF-8'
return yaml.safe_dump(value).encode('utf-8')
def build_swagger_20_swagger_schema_views(config):
settings = config.registry.settings
if settings.get('pyramid_swagger.dereference_served_schema'):
views = _build_dereferenced_swagger_20_schema_views(config)
else:
views = _build_swagger_20_schema_views(config)
return views
def _build_dereferenced_swagger_20_schema_views(config):
def view_for_swagger_schema(request):
settings = config.registry.settings
resolved_dict = settings.get('pyramid_swagger.schema20_resolved')
if not resolved_dict:
resolved_dict = settings['pyramid_swagger.schema20'].flattened_spec
settings['pyramid_swagger.schema20_resolved'] = resolved_dict
return resolved_dict
for schema_format in ['yaml', 'json']:
route_name = 'pyramid_swagger.swagger20.api_docs.{0}'\
.format(schema_format)
yield PyramidEndpoint(
path='/swagger.{0}'.format(schema_format),
view=view_for_swagger_schema,
route_name=route_name,
renderer=schema_format,
)
def _build_swagger_20_schema_views(config):
spec = config.registry.settings['pyramid_swagger.schema20']
walker = NodeWalkerForRefFiles()
all_files = walker.walk(spec)
file_map = {}
def view_for_swagger_schema(request):
_, ext = os.path.splitext(request.path)
ext = ext.lstrip('.')
base_path = config.registry.settings\
.get('pyramid_swagger.base_path_api_docs', '').rstrip('/')
key_path = request.path_info[len(base_path):]
actual_fname = file_map[key_path]
with spec.resolver.resolving(actual_fname) as spec_dict:
clean_response = strip_xscope(spec_dict)
ref_walker = NodeWalkerForCleaningRefs()
fixed_spec = ref_walker.walk(clean_response, ext)
return fixed_spec
for ref_fname in all_files:
ref_fname_parts = os.path.splitext(pathname2url(ref_fname))
for schema_format in ['yaml', 'json']:
route_name = 'pyramid_swagger.swagger20.api_docs.{0}.{1}'\
.format(ref_fname.replace('/', '.'), schema_format)
path = '/{0}.{1}'.format(ref_fname_parts[0], schema_format)
file_map[path] = ref_fname
yield PyramidEndpoint(
path=path,
route_name=route_name,
view=view_for_swagger_schema,
renderer=schema_format,
)