quay/endpoints/api/discovery.py

""" 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)
    path_swagger = {
      'name': fully_qualified_name(view_class),
      'path': swagger_path,
      '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['quay_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') or 'unnamed',
        'parameters': [],
      }

      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['internal'] = True

      if include_internal:
        requires_fresh_login = method_metadata(method, 'requires_fresh_login')
        if requires_fresh_login is not None:
          operation_swagger['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}]

      # TODO: 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 response_schema_name and not compact:
        models[response_schema_name] = view_class.schemas[response_schema_name]
        response_type = response_schema_name

      # 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]['tag']))

  if compact:
    return {'paths': paths}

  swagger_data = {
    'swagger': '2.0',
    'host': SERVER_HOSTNAME,
    'basePath': '/',
    'schemes': [
      PREFERRED_URL_SCHEME
    ],
    'info': {
      'version': 'v1',
      'title': 'Quay.io Frontend',
      '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 <a href="https://quay.io">Quay.io</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.ALL_SCOPES.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, args):
    """ List all of the API endpoints available in the swagger API format."""
    return swagger_route_data(args['internal'])