This repository has been archived on 2020-03-24. You can view files and clone it, but cannot push or open issues or pull requests.
quay/endpoints/api/discovery.py

305 lines
9.5 KiB
Python

""" API discovery information. """
import re
import logging
import sys
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
from collections import OrderedDict
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):
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 not compact:
parameter_info['description'] = description or ''
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()
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
}
if include_internal:
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
operation_swagger = {
'operationId': method_metadata(method, 'nickname'),
'parameters': [],
}
if operation_swagger['operationId'] is None:
continue
if not compact:
operation_swagger.update({
'description': method.__doc__.strip() if method.__doc__ else '',
'tags': [tag_name]
})
# 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'
},
'type': {
'type': 'string'
},
'detail': {
'type': 'string'
},
'title': {
'type': 'string'
}
},
'required': [
'status',
'type',
'title',
]
}
responses = {
'400': {
'description': 'Bad Request',
},
'401': {
'description': 'Session required',
},
'403': {
'description': 'Unauthorized access',
},
'404': {
'description': 'Not found',
},
}
for status, body in responses.items():
body['schema'] = {'$ref': '#/definitions/ApiError'}
if method_name == 'DELETE':
responses['204'] = {
'description': 'Deleted'
}
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}
swagger_data = {
'swagger': '2.0',
'host': SERVER_HOSTNAME,
'basePath': '/',
'schemes': [
PREFERRED_URL_SCHEME
],
'info': {
'version': 'v1',
'title': 'Quay Frontend',
'description': ('This API allows you to perform many of the operations required to work '
'with Quay repositories, users, and organizations. You can find out more '
'at <a href="https://quay.io">Quay</a>.'),
'termsOfService': 'https://quay.io/tos',
'contact': {
'email': 'support@quay.io'
}
},
'securityDefinitions': {
'oauth2_implicit': {
"type": "oauth2",
"flow": "implicit",
"authorizationUrl": "%s://%s/oauth/authorize" % (PREFERRED_URL_SCHEME, SERVER_HOSTNAME),
'scopes': {scope.scope:scope.description
for scope in scopes.app_scopes(app.config).values()},
},
},
'paths': paths,
'definitions': models,
'tags': tags
}
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, parsed_args):
""" List all of the API endpoints available in the swagger API format."""
return swagger_route_data(parsed_args['internal'])