252 lines
9.5 KiB
Python
252 lines
9.5 KiB
Python
import logging
|
|
import sys
|
|
from collections import OrderedDict
|
|
|
|
from config_app import app
|
|
from config_endpoints.api import method_metadata
|
|
from config_endpoints.common import fully_qualified_name, PARAM_REGEX, TYPE_CONVERTER
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def generate_route_data():
|
|
include_internal = True
|
|
compact = True
|
|
|
|
def swagger_parameter(name, description, kind='path', param_type='string', required=True,
|
|
enum=None, schema=None):
|
|
# https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#parameterObject
|
|
parameter_info = {
|
|
'name': name,
|
|
'in': kind,
|
|
'required': required
|
|
}
|
|
|
|
if schema:
|
|
parameter_info['schema'] = {
|
|
'$ref': '#/definitions/%s' % schema
|
|
}
|
|
else:
|
|
parameter_info['type'] = param_type
|
|
|
|
if enum is not None and len(list(enum)) > 0:
|
|
parameter_info['enum'] = list(enum)
|
|
|
|
return parameter_info
|
|
|
|
paths = {}
|
|
models = {}
|
|
tags = []
|
|
tags_added = set()
|
|
operation_ids = set()
|
|
|
|
for rule in app.url_map.iter_rules():
|
|
endpoint_method = app.view_functions[rule.endpoint]
|
|
|
|
# Verify that we have a view class for this API method.
|
|
if not 'view_class' in dir(endpoint_method):
|
|
continue
|
|
|
|
view_class = endpoint_method.view_class
|
|
|
|
# Hide the class if it is internal.
|
|
internal = method_metadata(view_class, 'internal')
|
|
if not include_internal and internal:
|
|
continue
|
|
|
|
# Build the tag.
|
|
parts = fully_qualified_name(view_class).split('.')
|
|
tag_name = parts[-2]
|
|
if not tag_name in tags_added:
|
|
tags_added.add(tag_name)
|
|
tags.append({
|
|
'name': tag_name,
|
|
'description': (sys.modules[view_class.__module__].__doc__ or '').strip()
|
|
})
|
|
|
|
# Build the Swagger data for the path.
|
|
swagger_path = PARAM_REGEX.sub(r'{\2}', rule.rule)
|
|
full_name = fully_qualified_name(view_class)
|
|
path_swagger = {
|
|
'x-name': full_name,
|
|
'x-path': swagger_path,
|
|
'x-tag': tag_name
|
|
}
|
|
|
|
related_user_res = method_metadata(view_class, 'related_user_resource')
|
|
if related_user_res is not None:
|
|
path_swagger['x-user-related'] = fully_qualified_name(related_user_res)
|
|
|
|
paths[swagger_path] = path_swagger
|
|
|
|
# Add any global path parameters.
|
|
param_data_map = view_class.__api_path_params if '__api_path_params' in dir(view_class) else {}
|
|
if param_data_map:
|
|
path_parameters_swagger = []
|
|
for path_parameter in param_data_map:
|
|
description = param_data_map[path_parameter].get('description')
|
|
path_parameters_swagger.append(swagger_parameter(path_parameter, description))
|
|
|
|
path_swagger['parameters'] = path_parameters_swagger
|
|
|
|
# Add the individual HTTP operations.
|
|
method_names = list(rule.methods.difference(['HEAD', 'OPTIONS']))
|
|
for method_name in method_names:
|
|
# https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#operation-object
|
|
method = getattr(view_class, method_name.lower(), None)
|
|
if method is None:
|
|
logger.debug('Unable to find method for %s in class %s', method_name, view_class)
|
|
continue
|
|
|
|
operationId = method_metadata(method, 'nickname')
|
|
operation_swagger = {
|
|
'operationId': operationId,
|
|
'parameters': [],
|
|
}
|
|
|
|
if operationId is None:
|
|
continue
|
|
|
|
if operationId in operation_ids:
|
|
raise Exception('Duplicate operation Id: %s' % operationId)
|
|
|
|
operation_ids.add(operationId)
|
|
|
|
# Mark the method as internal.
|
|
internal = method_metadata(method, 'internal')
|
|
if internal is not None:
|
|
operation_swagger['x-internal'] = True
|
|
|
|
if include_internal:
|
|
requires_fresh_login = method_metadata(method, 'requires_fresh_login')
|
|
if requires_fresh_login is not None:
|
|
operation_swagger['x-requires-fresh-login'] = True
|
|
|
|
# Add the path parameters.
|
|
if rule.arguments:
|
|
for path_parameter in rule.arguments:
|
|
description = param_data_map.get(path_parameter, {}).get('description')
|
|
operation_swagger['parameters'].append(swagger_parameter(path_parameter, description))
|
|
|
|
# Add the query parameters.
|
|
if '__api_query_params' in dir(method):
|
|
for query_parameter_info in method.__api_query_params:
|
|
name = query_parameter_info['name']
|
|
description = query_parameter_info['help']
|
|
param_type = TYPE_CONVERTER[query_parameter_info['type']]
|
|
required = query_parameter_info['required']
|
|
|
|
operation_swagger['parameters'].append(
|
|
swagger_parameter(name, description, kind='query',
|
|
param_type=param_type,
|
|
required=required,
|
|
enum=query_parameter_info['choices']))
|
|
|
|
# Add the OAuth security block.
|
|
# https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#securityRequirementObject
|
|
scope = method_metadata(method, 'oauth2_scope')
|
|
if scope and not compact:
|
|
operation_swagger['security'] = [{'oauth2_implicit': [scope.scope]}]
|
|
|
|
# Add the responses block.
|
|
# https://github.com/swagger-api/swagger-spec/blob/master/versions/2.0.md#responsesObject
|
|
response_schema_name = method_metadata(method, 'response_schema')
|
|
if not compact:
|
|
if response_schema_name:
|
|
models[response_schema_name] = view_class.schemas[response_schema_name]
|
|
|
|
models['ApiError'] = {
|
|
'type': 'object',
|
|
'properties': {
|
|
'status': {
|
|
'type': 'integer',
|
|
'description': 'Status code of the response.'
|
|
},
|
|
'type': {
|
|
'type': 'string',
|
|
'description': 'Reference to the type of the error.'
|
|
},
|
|
'detail': {
|
|
'type': 'string',
|
|
'description': 'Details about the specific instance of the error.'
|
|
},
|
|
'title': {
|
|
'type': 'string',
|
|
'description': 'Unique error code to identify the type of error.'
|
|
},
|
|
'error_message': {
|
|
'type': 'string',
|
|
'description': 'Deprecated; alias for detail'
|
|
},
|
|
'error_type': {
|
|
'type': 'string',
|
|
'description': 'Deprecated; alias for detail'
|
|
}
|
|
},
|
|
'required': [
|
|
'status',
|
|
'type',
|
|
'title',
|
|
]
|
|
}
|
|
|
|
responses = {
|
|
'400': {
|
|
'description': 'Bad Request',
|
|
},
|
|
|
|
'401': {
|
|
'description': 'Session required',
|
|
},
|
|
|
|
'403': {
|
|
'description': 'Unauthorized access',
|
|
},
|
|
|
|
'404': {
|
|
'description': 'Not found',
|
|
},
|
|
}
|
|
|
|
for _, body in responses.items():
|
|
body['schema'] = {'$ref': '#/definitions/ApiError'}
|
|
|
|
if method_name == 'DELETE':
|
|
responses['204'] = {
|
|
'description': 'Deleted'
|
|
}
|
|
elif method_name == 'POST':
|
|
responses['201'] = {
|
|
'description': 'Successful creation'
|
|
}
|
|
else:
|
|
responses['200'] = {
|
|
'description': 'Successful invocation'
|
|
}
|
|
|
|
if response_schema_name:
|
|
responses['200']['schema'] = {
|
|
'$ref': '#/definitions/%s' % response_schema_name
|
|
}
|
|
|
|
operation_swagger['responses'] = responses
|
|
|
|
# Add the request block.
|
|
request_schema_name = method_metadata(method, 'request_schema')
|
|
if request_schema_name and not compact:
|
|
models[request_schema_name] = view_class.schemas[request_schema_name]
|
|
|
|
operation_swagger['parameters'].append(
|
|
swagger_parameter('body', 'Request body contents.', kind='body',
|
|
schema=request_schema_name))
|
|
|
|
# Add the operation to the parent path.
|
|
if not internal or (internal and include_internal):
|
|
path_swagger[method_name.lower()] = operation_swagger
|
|
|
|
tags.sort(key=lambda t: t['name'])
|
|
paths = OrderedDict(sorted(paths.items(), key=lambda p: p[1]['x-tag']))
|
|
|
|
if compact:
|
|
return {'paths': paths}
|