mirror of
https://github.com/asterisk/asterisk.git
synced 2025-10-29 07:24:55 +00:00
255 lines
8.9 KiB
Python
255 lines
8.9 KiB
Python
#
|
|
# Asterisk -- An open source telephony toolkit.
|
|
#
|
|
# Copyright (C) 2013, Digium, Inc.
|
|
#
|
|
# David M. Lee, II <dlee@digium.com>
|
|
#
|
|
# See http://www.asterisk.org for more information about
|
|
# the Asterisk project. Please do not directly contact
|
|
# any of the maintainers of this project for assistance;
|
|
# the project provides a web site, mailing lists and IRC
|
|
# channels for your use.
|
|
#
|
|
# This program is free software, distributed under the terms of
|
|
# the GNU General Public License Version 2. See the LICENSE file
|
|
# at the top of the source tree.
|
|
#
|
|
|
|
"""Implementation of SwaggerPostProcessor which adds fields needed to generate
|
|
Asterisk RESTful HTTP binding code.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
|
|
from swagger_model import Stringify, SwaggerError, SwaggerPostProcessor
|
|
|
|
try:
|
|
from collections import OrderedDict
|
|
except ImportError:
|
|
from odict import OrderedDict
|
|
|
|
|
|
def simple_name(name):
|
|
"""Removes the {markers} from a path segement.
|
|
|
|
@param name: Swagger path segement, with {pathVar} markers.
|
|
"""
|
|
if name.startswith('{') and name.endswith('}'):
|
|
return name[1:-1]
|
|
return name
|
|
|
|
|
|
def wikify(str):
|
|
"""Escapes a string for the wiki.
|
|
|
|
@param str: String to escape
|
|
"""
|
|
# Replace all line breaks with line feeds
|
|
str = re.sub(r'<br\s*/?>', '\n', str)
|
|
return re.sub(r'([{}\[\]])', r'\\\1', str)
|
|
|
|
|
|
def snakify(name):
|
|
"""Helper to take a camelCase or dash-seperated name and make it
|
|
snake_case.
|
|
"""
|
|
r = ''
|
|
prior_lower = False
|
|
for c in name:
|
|
if c.isupper() and prior_lower:
|
|
r += "_"
|
|
if c == '-':
|
|
c = '_'
|
|
prior_lower = c.islower()
|
|
r += c.lower()
|
|
return r
|
|
|
|
|
|
class PathSegment(Stringify):
|
|
"""Tree representation of a Swagger API declaration.
|
|
"""
|
|
def __init__(self, name, parent):
|
|
"""Ctor.
|
|
|
|
@param name: Name of this path segment. May have {pathVar} markers.
|
|
@param parent: Parent PathSegment.
|
|
"""
|
|
#: Segment name, with {pathVar} markers removed
|
|
self.name = simple_name(name)
|
|
#: True if segment is a {pathVar}, else None.
|
|
self.is_wildcard = None
|
|
#: Underscore seperated name all ancestor segments
|
|
self.full_name = None
|
|
#: Dictionary of child PathSegements
|
|
self.__children = OrderedDict()
|
|
#: List of operations on this segement
|
|
self.operations = []
|
|
|
|
if self.name != name:
|
|
self.is_wildcard = True
|
|
|
|
if not self.name:
|
|
assert(not parent)
|
|
self.full_name = ''
|
|
if not parent or not parent.name:
|
|
self.full_name = name
|
|
else:
|
|
self.full_name = "%s_%s" % (parent.full_name, self.name)
|
|
|
|
def get_child(self, path):
|
|
"""Walks descendants to get path, creating it if necessary.
|
|
|
|
@param path: List of path names.
|
|
@return: PageSegment corresponding to path.
|
|
"""
|
|
assert simple_name(path[0]) == self.name
|
|
if (len(path) == 1):
|
|
return self
|
|
child = self.__children.get(path[1])
|
|
if not child:
|
|
child = PathSegment(path[1], self)
|
|
self.__children[path[1]] = child
|
|
return child.get_child(path[1:])
|
|
|
|
def children(self):
|
|
"""Gets list of children.
|
|
"""
|
|
return self.__children.values()
|
|
|
|
def num_children(self):
|
|
"""Gets count of children.
|
|
"""
|
|
return len(self.__children)
|
|
|
|
|
|
class AsteriskProcessor(SwaggerPostProcessor):
|
|
"""A SwaggerPostProcessor which adds fields needed to generate Asterisk
|
|
RESTful HTTP binding code.
|
|
"""
|
|
|
|
#: How Swagger types map to C.
|
|
type_mapping = {
|
|
'string': 'const char *',
|
|
'boolean': 'int',
|
|
'number': 'int',
|
|
'int': 'int',
|
|
'long': 'long',
|
|
'double': 'double',
|
|
'float': 'float',
|
|
}
|
|
|
|
#: String conversion functions for string to C type.
|
|
convert_mapping = {
|
|
'string': '',
|
|
'int': 'atoi',
|
|
'long': 'atol',
|
|
'double': 'atof',
|
|
'boolean': 'ast_true',
|
|
}
|
|
|
|
#: JSON conversion functions
|
|
json_convert_mapping = {
|
|
'string': 'ast_json_string_get',
|
|
'int': 'ast_json_integer_get',
|
|
'long': 'ast_json_integer_get',
|
|
'double': 'ast_json_real_get',
|
|
'boolean': 'ast_json_is_true',
|
|
}
|
|
|
|
def __init__(self, wiki_prefix):
|
|
self.wiki_prefix = wiki_prefix
|
|
|
|
def process_resource_api(self, resource_api, context):
|
|
resource_api.wiki_prefix = self.wiki_prefix
|
|
# Derive a resource name from the API declaration's filename
|
|
resource_api.name = re.sub('\..*', '',
|
|
os.path.basename(resource_api.path))
|
|
# Now in all caps, for include guard
|
|
resource_api.name_caps = resource_api.name.upper()
|
|
resource_api.name_title = resource_api.name.capitalize()
|
|
resource_api.c_name = snakify(resource_api.name)
|
|
# Construct the PathSegement tree for the API.
|
|
if resource_api.api_declaration:
|
|
resource_api.root_path = PathSegment('', None)
|
|
for api in resource_api.api_declaration.apis:
|
|
segment = resource_api.root_path.get_child(api.path.split('/'))
|
|
for operation in api.operations:
|
|
segment.operations.append(operation)
|
|
api.full_name = segment.full_name
|
|
|
|
# Since every API path should start with /[resource], root should
|
|
# have exactly one child.
|
|
if resource_api.root_path.num_children() != 1:
|
|
raise SwaggerError(
|
|
"Should not mix resources in one API declaration", context)
|
|
# root_path isn't needed any more
|
|
resource_api.root_path = list(resource_api.root_path.children())[0]
|
|
if resource_api.name != resource_api.root_path.name:
|
|
raise SwaggerError(
|
|
"API declaration name should match", context)
|
|
resource_api.root_full_name = resource_api.root_path.full_name
|
|
|
|
def process_api(self, api, context):
|
|
api.wiki_path = wikify(api.path)
|
|
|
|
def process_operation(self, operation, context):
|
|
# Nicknames are camelCase, Asterisk coding is snake case
|
|
operation.c_nickname = snakify(operation.nickname)
|
|
operation.c_http_method = 'AST_HTTP_' + operation.http_method
|
|
if not operation.summary.endswith("."):
|
|
raise SwaggerError("Summary should end with .", context)
|
|
operation.wiki_summary = wikify(operation.summary or "")
|
|
operation.wiki_notes = wikify(operation.notes or "")
|
|
for error_response in operation.error_responses:
|
|
error_response.wiki_reason = wikify(error_response.reason or "")
|
|
operation.parse_body = (operation.body_parameter or operation.has_query_parameters) and True
|
|
|
|
def process_parameter(self, parameter, context):
|
|
if parameter.param_type == 'body':
|
|
parameter.is_body_parameter = True;
|
|
parameter.c_data_type = 'struct ast_json *'
|
|
else:
|
|
parameter.is_body_parameter = False;
|
|
if not parameter.data_type in self.type_mapping:
|
|
raise SwaggerError(
|
|
"Invalid parameter type %s" % parameter.data_type, context)
|
|
# Type conversions
|
|
parameter.c_data_type = self.type_mapping[parameter.data_type]
|
|
parameter.c_convert = self.convert_mapping[parameter.data_type]
|
|
parameter.json_convert = self.json_convert_mapping[parameter.data_type]
|
|
|
|
# Parameter names are camelcase, Asterisk convention is snake case
|
|
parameter.c_name = snakify(parameter.name)
|
|
# You shouldn't put a space between 'char *' and the variable
|
|
if parameter.c_data_type.endswith('*'):
|
|
parameter.c_space = ''
|
|
else:
|
|
parameter.c_space = ' '
|
|
parameter.wiki_description = wikify(parameter.description)
|
|
if parameter.allowable_values:
|
|
parameter.wiki_allowable_values = parameter.allowable_values.to_wiki()
|
|
else:
|
|
parameter.wiki_allowable_values = None
|
|
|
|
def process_model(self, model, context):
|
|
model.description_dox = model.description.replace('\n', '\n * ')
|
|
model.description_dox = re.sub(' *\n', '\n', model.description_dox)
|
|
model.wiki_description = wikify(model.description)
|
|
model.c_id = snakify(model.id)
|
|
return model
|
|
|
|
def process_property(self, prop, context):
|
|
if "-" in prop.name:
|
|
raise SwaggerError("Property names cannot have dashes", context)
|
|
if prop.name != prop.name.lower():
|
|
raise SwaggerError("Property name should be all lowercase",
|
|
context)
|
|
prop.wiki_description = wikify(prop.description)
|
|
|
|
def process_type(self, swagger_type, context):
|
|
swagger_type.c_name = snakify(swagger_type.name)
|
|
swagger_type.c_singular_name = snakify(swagger_type.singular_name)
|
|
swagger_type.wiki_name = wikify(swagger_type.name)
|