Add channel events for res_stasis apps
[asterisk/asterisk.git] / rest-api-templates / swagger_model.py
1
2 # Asterisk -- An open source telephony toolkit.
3 #
4 # Copyright (C) 2013, Digium, Inc.
5 #
6 # David M. Lee, II <dlee@digium.com>
7 #
8 # See http://www.asterisk.org for more information about
9 # the Asterisk project. Please do not directly contact
10 # any of the maintainers of this project for assistance;
11 # the project provides a web site, mailing lists and IRC
12 # channels for your use.
13 #
14 # This program is free software, distributed under the terms of
15 # the GNU General Public License Version 2. See the LICENSE file
16 # at the top of the source tree.
17 #
18
19 """Swagger data model objects.
20
21 These objects should map directly to the Swagger api-docs, without a lot of
22 additional fields. In the process of translation, it should also validate the
23 model for consistency against the Swagger spec (i.e., fail if fields are
24 missing, or have incorrect values).
25
26 See https://github.com/wordnik/swagger-core/wiki/API-Declaration for the spec.
27 """
28
29 import json
30 import os.path
31 import pprint
32 import sys
33 import traceback
34
35 try:
36     from collections import OrderedDict
37 except ImportError:
38     from odict import OrderedDict
39
40
41 SWAGGER_VERSION = "1.1"
42
43
44 class SwaggerError(Exception):
45     """Raised when an error is encountered mapping the JSON objects into the
46     model.
47     """
48
49     def __init__(self, msg, context, cause=None):
50         """Ctor.
51
52         @param msg: String message for the error.
53         @param context: Array of strings for current context in the API.
54         @param cause: Optional exception that caused this one.
55         """
56         super(Exception, self).__init__(msg, context, cause)
57
58
59 class SwaggerPostProcessor(object):
60     """Post processing interface for model objects. This processor can add
61     fields to model objects for additional information to use in the
62     templates.
63     """
64     def process_api(self, resource_api, context):
65         """Post process a ResourceApi object.
66
67         @param resource_api: ResourceApi object.
68         @param context: Current context in the API.
69         """
70         pass
71
72     def process_operation(self, operation, context):
73         """Post process a Operation object.
74
75         @param operation: Operation object.
76         @param context: Current context in the API.
77         """
78         pass
79
80     def process_parameter(self, parameter, context):
81         """Post process a Parameter object.
82
83         @param parameter: Parameter object.
84         @param context: Current context in the API.
85         """
86         pass
87
88
89 class Stringify(object):
90     """Simple mix-in to make the repr of the model classes more meaningful.
91     """
92     def __repr__(self):
93         return "%s(%s)" % (self.__class__, pprint.saferepr(self.__dict__))
94
95
96 class AllowableRange(Stringify):
97     """Model of a allowableValues of type RANGE
98
99     See https://github.com/wordnik/swagger-core/wiki/datatypes#complex-types
100     """
101     def __init__(self, min_value, max_value):
102         self.min_value = min_value
103         self.max_value = max_value
104
105
106 class AllowableList(Stringify):
107     """Model of a allowableValues of type LIST
108
109     See https://github.com/wordnik/swagger-core/wiki/datatypes#complex-types
110     """
111     def __init__(self, values):
112         self.values = values
113
114
115 def load_allowable_values(json, context):
116     """Parse a JSON allowableValues object.
117
118     This returns None, AllowableList or AllowableRange, depending on the
119     valueType in the JSON. If the valueType is not recognized, a SwaggerError
120     is raised.
121     """
122     if not json:
123         return None
124
125     if not 'valueType' in json:
126         raise SwaggerError("Missing valueType field", context)
127
128     value_type = json['valueType']
129
130     if value_type == 'RANGE':
131         if not 'min' in json:
132             raise SwaggerError("Missing field min", context)
133         if not 'max' in json:
134             raise SwaggerError("Missing field max", context)
135         return AllowableRange(json['min'], json['max'])
136     if value_type == 'LIST':
137         if not 'values' in json:
138             raise SwaggerError("Missing field values", context)
139         return AllowableList(json['values'])
140     raise SwaggerError("Unkown valueType %s" % value_type, context)
141
142
143 class Parameter(Stringify):
144     """Model of an operation's parameter.
145
146     See https://github.com/wordnik/swagger-core/wiki/parameters
147     """
148
149     required_fields = ['name', 'paramType', 'dataType']
150
151     def __init__(self):
152         self.param_type = None
153         self.name = None
154         self.description = None
155         self.data_type = None
156         self.required = None
157         self.allowable_values = None
158         self.allow_multiple = None
159
160     def load(self, parameter_json, processor, context):
161         context = add_context(context, parameter_json, 'name')
162         validate_required_fields(parameter_json, self.required_fields, context)
163         self.name = parameter_json.get('name')
164         self.param_type = parameter_json.get('paramType')
165         self.description = parameter_json.get('description') or ''
166         self.data_type = parameter_json.get('dataType')
167         self.required = parameter_json.get('required') or False
168         self.allowable_values = load_allowable_values(
169             parameter_json.get('allowableValues'), context)
170         self.allow_multiple = parameter_json.get('allowMultiple') or False
171         processor.process_parameter(self, context)
172         return self
173
174     def is_type(self, other_type):
175         return self.param_type == other_type
176
177
178 class ErrorResponse(Stringify):
179     """Model of an error response.
180
181     See https://github.com/wordnik/swagger-core/wiki/errors
182     """
183
184     required_fields = ['code', 'reason']
185
186     def __init__(self):
187         self.code = None
188         self.reason = None
189
190     def load(self, err_json, processor, context):
191         context = add_context(context, err_json, 'code')
192         validate_required_fields(err_json, self.required_fields, context)
193         self.code = err_json.get('code')
194         self.reason = err_json.get('reason')
195         return self
196
197
198 class Operation(Stringify):
199     """Model of an operation on an API
200
201     See https://github.com/wordnik/swagger-core/wiki/API-Declaration#apis
202     """
203
204     required_fields = ['httpMethod', 'nickname', 'responseClass', 'summary']
205
206     def __init__(self):
207         self.http_method = None
208         self.nickname = None
209         self.response_class = None
210         self.parameters = []
211         self.summary = None
212         self.notes = None
213         self.error_responses = []
214
215     def load(self, op_json, processor, context):
216         context = add_context(context, op_json, 'nickname')
217         validate_required_fields(op_json, self.required_fields, context)
218         self.http_method = op_json.get('httpMethod')
219         self.nickname = op_json.get('nickname')
220         self.response_class = op_json.get('responseClass')
221         params_json = op_json.get('parameters') or []
222         self.parameters = [
223             Parameter().load(j, processor, context) for j in params_json]
224         self.query_parameters = [
225             p for p in self.parameters if p.is_type('query')]
226         self.has_query_parameters = self.query_parameters and True
227         self.path_parameters = [
228             p for p in self.parameters if p.is_type('path')]
229         self.has_path_parameters = self.path_parameters and True
230         self.header_parameters = [
231             p for p in self.parameters if p.is_type('header')]
232         self.has_header_parameters = self.header_parameters and True
233         self.has_parameters = self.has_query_parameters or \
234             self.has_path_parameters or self.has_header_parameters
235         self.summary = op_json.get('summary')
236         self.notes = op_json.get('notes')
237         err_json = op_json.get('errorResponses') or []
238         self.error_responses = [
239             ErrorResponse().load(j, processor, context) for j in err_json]
240         processor.process_operation(self, context)
241         return self
242
243
244 class Api(Stringify):
245     """Model of a single API in an API declaration.
246
247     See https://github.com/wordnik/swagger-core/wiki/API-Declaration
248     """
249
250     required_fields = ['path', 'operations']
251
252     def __init__(self,):
253         self.path = None
254         self.description = None
255         self.operations = []
256
257     def load(self, api_json, processor, context):
258         context = add_context(context, api_json, 'path')
259         validate_required_fields(api_json, self.required_fields, context)
260         self.path = api_json.get('path')
261         self.description = api_json.get('description')
262         op_json = api_json.get('operations')
263         self.operations = [
264             Operation().load(j, processor, context) for j in op_json]
265         return self
266
267
268 class Property(Stringify):
269     """Model of a Swagger property.
270
271     See https://github.com/wordnik/swagger-core/wiki/datatypes
272     """
273
274     required_fields = ['type']
275
276     def __init__(self, name):
277         self.name = name
278         self.type = None
279         self.description = None
280         self.required = None
281
282     def load(self, property_json, processor, context):
283         validate_required_fields(property_json, self.required_fields, context)
284         self.type = property_json.get('type')
285         self.description = property_json.get('description') or ''
286         self.required = property_json.get('required') or False
287         return self
288
289
290 class Model(Stringify):
291     """Model of a Swagger model.
292
293     See https://github.com/wordnik/swagger-core/wiki/datatypes
294     """
295
296     def __init__(self):
297         self.id = None
298         self.notes = None
299         self.description = None
300         self.properties = None
301
302     def load(self, id, model_json, processor, context):
303         context = add_context(context, model_json, 'id')
304         # This arrangement is required by the Swagger API spec
305         self.id = model_json.get('id')
306         if id != self.id:
307             raise SwaggerError("Model id doesn't match name", c)
308         self.description = model_json.get('description')
309         props = model_json.get('properties').items() or []
310         self.properties = [
311             Property(k).load(j, processor, context) for (k, j) in props]
312         return self
313
314
315 class ApiDeclaration(Stringify):
316     """Model class for an API Declaration.
317
318     See https://github.com/wordnik/swagger-core/wiki/API-Declaration
319     """
320
321     required_fields = [
322         'swaggerVersion', '_author', '_copyright', 'apiVersion', 'basePath',
323         'resourcePath', 'apis', 'models'
324     ]
325
326     def __init__(self):
327         self.swagger_version = None
328         self.author = None
329         self.copyright = None
330         self.api_version = None
331         self.base_path = None
332         self.resource_path = None
333         self.apis = []
334         self.models = []
335
336     def load_file(self, api_declaration_file, processor, context=[]):
337         context = context + [api_declaration_file]
338         try:
339             return self.__load_file(api_declaration_file, processor, context)
340         except SwaggerError:
341             raise
342         except Exception as e:
343             print >> sys.stderr, "Error: ", traceback.format_exc()
344             raise SwaggerError(
345                 "Error loading %s" % api_declaration_file, context, e)
346
347     def __load_file(self, api_declaration_file, processor, context):
348         with open(api_declaration_file) as fp:
349             self.load(json.load(fp), processor, context)
350
351         expected_resource_path = '/api-docs/' + \
352             os.path.basename(api_declaration_file) \
353             .replace(".json", ".{format}")
354
355         if self.resource_path != expected_resource_path:
356             print "%s != %s" % (self.resource_path, expected_resource_path)
357             raise SwaggerError("resourcePath has incorrect value", context)
358
359         return self
360
361     def load(self, api_decl_json, processor, context):
362         """Loads a resource from a single Swagger resource.json file.
363         """
364         # If the version doesn't match, all bets are off.
365         self.swagger_version = api_decl_json.get('swaggerVersion')
366         if self.swagger_version != SWAGGER_VERSION:
367             raise SwaggerError(
368                 "Unsupported Swagger version %s" % swagger_version, context)
369
370         validate_required_fields(api_decl_json, self.required_fields, context)
371
372         self.author = api_decl_json.get('_author')
373         self.copyright = api_decl_json.get('_copyright')
374         self.api_version = api_decl_json.get('apiVersion')
375         self.base_path = api_decl_json.get('basePath')
376         self.resource_path = api_decl_json.get('resourcePath')
377         api_json = api_decl_json.get('apis') or []
378         self.apis = [
379             Api().load(j, processor, context) for j in api_json]
380         models = api_decl_json.get('models').items() or []
381         self.models = [
382             Model().load(k, j, processor, context) for (k, j) in models]
383
384         return self
385
386
387 class ResourceApi(Stringify):
388     """Model of an API listing in the resources.json file.
389     """
390
391     required_fields = ['path', 'description']
392
393     def __init__(self):
394         self.path = None
395         self.description = None
396         self.api_declaration = None
397
398     def load(self, api_json, processor, context):
399         context = add_context(context, api_json, 'path')
400         validate_required_fields(api_json, self.required_fields, context)
401         self.path = api_json['path']
402         self.description = api_json['description']
403
404         if not self.path or self.path[0] != '/':
405             raise SwaggerError("Path must start with /", context)
406         processor.process_api(self, context)
407         return self
408
409     def load_api_declaration(self, base_dir, processor):
410         self.file = (base_dir + self.path).replace('{format}', 'json')
411         self.api_declaration = ApiDeclaration().load_file(self.file, processor)
412         processor.process_api(self, [self.file])
413
414
415 class ResourceListing(Stringify):
416     """Model of Swagger's resources.json file.
417     """
418
419     required_fields = ['apiVersion', 'basePath', 'apis']
420
421     def __init__(self):
422         self.swagger_version = None
423         self.api_version = None
424         self.base_path = None
425         self.apis = None
426
427     def load_file(self, resource_file, processor):
428         context = [resource_file]
429         try:
430             return self.__load_file(resource_file, processor, context)
431         except SwaggerError:
432             raise
433         except Exception as e:
434             print >> sys.stderr, "Error: ", traceback.format_exc()
435             raise SwaggerError(
436                 "Error loading %s" % resource_file, context, e)
437
438     def __load_file(self, resource_file, processor, context):
439         with open(resource_file) as fp:
440             return self.load(json.load(fp), processor, context)
441
442     def load(self, resources_json, processor, context):
443         # If the version doesn't match, all bets are off.
444         self.swagger_version = resources_json.get('swaggerVersion')
445         if self.swagger_version != SWAGGER_VERSION:
446             raise SwaggerError(
447                 "Unsupported Swagger version %s" % swagger_version, context)
448
449         validate_required_fields(resources_json, self.required_fields, context)
450         self.api_version = resources_json['apiVersion']
451         self.base_path = resources_json['basePath']
452         apis_json = resources_json['apis']
453         self.apis = [
454             ResourceApi().load(j, processor, context) for j in apis_json]
455         return self
456
457
458 def validate_required_fields(json, required_fields, context):
459     """Checks a JSON object for a set of required fields.
460
461     If any required field is missing, a SwaggerError is raised.
462
463     @param json: JSON object to check.
464     @param required_fields: List of required fields.
465     @param context: Current context in the API.
466     """
467     missing_fields = [f for f in required_fields if not f in json]
468
469     if missing_fields:
470         raise SwaggerError(
471             "Missing fields: %s" % ', '.join(missing_fields), context)
472
473
474 def add_context(context, json, id_field):
475     """Returns a new context with a new item added to it.
476
477     @param context: Old context.
478     @param json: Current JSON object.
479     @param id_field: Field identifying this object.
480     @return New context with additional item.
481     """
482     if not id_field in json:
483         raise SwaggerError("Missing id_field: %s" % id_field, context)
484     return context + ['%s=%s' % (id_field, str(json[id_field]))]