import re import logging from flask.ext.restful import reqparse from endpoints.api import (ApiResource, resource, method_metadata, nickname, truthy_bool, parse_args, query_param) 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', } PREFERRED_URL_SCHEME = app.config['PREFERRED_URL_SCHEME'] SERVER_HOSTNAME = app.config['SERVER_HOSTNAME'] def fully_qualified_name(method_view_class): return '%s.%s' % (method_view_class.__module__, method_view_class.__name__) def swagger_route_data(include_internal=False, compact=False): 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 param_data_map = {} if '__api_path_params' in dir(view_class): param_data_map = view_class.__api_path_params 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_data_map.get(param, {'description': ''})['description'], 'required': True, }) if method is None: logger.debug('Unable to find method for %s in class %s', method_name, view_class) else: 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') } if not compact: new_operation.update({ 'type': 'void', 'summary': method.__doc__.strip() if method.__doc__ else '', 'parameters': parameters, }) scope = method_metadata(method, 'oauth2_scope') if scope and not compact: new_operation['authorizations'] = { 'oauth2': [ { 'scope': scope.scope } ], } internal = method_metadata(method, 'internal') if internal is not None: new_operation['internal'] = True if not internal or (internal and include_internal): operations.append(new_operation) swagger_path = PARAM_REGEX.sub(r'{\2}', rule.rule) new_resource = { 'path': swagger_path, 'description': view_class.__doc__.strip() if view_class.__doc__ else "", 'operations': operations, 'name': fully_qualified_name(view_class), } related_user_res = method_metadata(view_class, 'related_user_resource') if related_user_res is not None: new_resource['quayUserRelated'] = fully_qualified_name(related_user_res) internal = method_metadata(view_class, 'internal') if internal is not None: new_resource['internal'] = True if not internal or (internal and include_internal): apis.append(new_resource) # If compact form was requested, simply return the APIs. if compact: return {'apis': apis} swagger_data = { 'apiVersion': 'v1', 'swaggerVersion': '1.2', 'basePath': '%s://%s' % (PREFERRED_URL_SCHEME, SERVER_HOSTNAME), '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': [scope._asdict() for scope in scopes.ALL_SCOPES.values()], 'grantTypes': { "implicit": { "tokenName": "access_token", "loginEndpoint": { "url": "%s://%s/oauth/authorize" % (PREFERRED_URL_SCHEME, SERVER_HOSTNAME), }, }, }, }, }, 'apis': apis, 'models': models, } return swagger_data @resource('/v1/discovery') class DiscoveryResource(ApiResource): """Ability to inspect the API for usage information and documentation.""" @parse_args @query_param('internal', 'Whether to include internal APIs.', type=truthy_bool, default=False) @nickname('discovery') def get(self, args): """ List all of the API endpoints available in the swagger API format.""" return swagger_route_data(args['internal'])