import re import logging from flask.ext.restful import reqparse from endpoints.api import ApiResource, resource, method_metadata, nickname, truthy_bool from app import app from auth import scopes logger = logging.getLogger(__name__) PARAM_REGEX = re.compile(r'<([\w]+:)?([\w]+)>') TYPE_CONVERTER = { truthy_bool: 'boolean', str: 'string', basestring: 'string', reqparse.text_type: 'string', int: 'integer', } def swagger_route_data(): apis = [] models = {} for rule in app.url_map.iter_rules(): endpoint_method = app.view_functions[rule.endpoint] if 'view_class' in dir(endpoint_method): view_class = endpoint_method.view_class operations = [] method_names = list(rule.methods.difference(['HEAD', 'OPTIONS'])) for method_name in method_names: method = getattr(view_class, method_name.lower(), None) parameters = [] for param in rule.arguments: parameters.append({ 'paramType': 'path', 'name': param, 'dataType': 'string', 'description': 'Param description.', 'required': True, }) if method is not None: req_schema_name = method_metadata(method, 'request_schema') if req_schema_name: parameters.append({ 'paramType': 'body', 'name': 'body', 'description': 'Request body contents.', 'dataType': req_schema_name, 'required': True, }) schema = view_class.schemas[req_schema_name] models[req_schema_name] = schema if '__api_query_params' in dir(method): for param_spec in method.__api_query_params: new_param = { 'paramType': 'query', 'name': param_spec['name'], 'description': param_spec['help'], 'dataType': TYPE_CONVERTER[param_spec['type']], 'required': param_spec['required'], } if len(param_spec['choices']) > 0: new_param['enum'] = list(param_spec['choices']) parameters.append(new_param) new_operation = { 'method': method_name, 'nickname': method_metadata(method, 'nickname'), 'type': 'void', 'summary': method.__doc__ if method.__doc__ else '', 'parameters': parameters, } scope = method_metadata(method, 'oauth2_scope') if scope: new_operation['authorizations'] = { 'oauth2': [scope] } operations.append(new_operation) swagger_path = PARAM_REGEX.sub(r'{\2}', rule.rule) apis.append({ 'path': swagger_path, 'description': view_class.__doc__ if view_class.__doc__ else "", 'operations': operations, }) swagger_data = { 'apiVersion': 'v1', 'swaggerVersion': '1.2', 'basePath': 'http://ci.devtable.com:5000', 'resourcePath': '/', 'info': { 'title': 'Quay.io API', 'description': ('This API allows you to perform many of the operations required to work ' 'with Quay.io repositories, users, and organizations. You can find out more ' 'at Quay.io.'), 'termsOfServiceUrl': 'https://quay.io/tos', 'contact': 'support@quay.io', }, 'authorizations': { 'oauth2': { 'scopes': list(scopes.ALL_SCOPES.values()), 'grantTypes': { "implicit": { "tokenName": "access_token", "loginEndpoint": { "url": "http://ci.devtable.com:5000/oauth/authorize", }, }, }, }, }, 'apis': apis, 'models': models, } return swagger_data @resource('/v1/discovery') class DiscoveryResource(ApiResource): """Ability to inspect the API for usage information and documentation.""" @nickname('discovery') def get(self): """ List all of the API endpoints available in the swagger API format.""" return swagger_route_data()