quay/config_app/config_endpoints/common.py

import logging
import os
import re
import sys
from collections import OrderedDict
from cachetools import lru_cache

from flask import make_response, render_template
from flask_restful import reqparse

from config_app.config_endpoints.api import method_metadata
from config_app.app import app


def truthy_bool(param):
    return param not in {False, 'false', 'False', '0', 'FALSE', '', 'null'}


DEFAULT_JS_BUNDLE_NAME = 'configapp'
PARAM_REGEX = re.compile(r'<([^:>]+:)*([\w]+)>')
logger = logging.getLogger(__name__)
TYPE_CONVERTER = {
    truthy_bool: 'boolean',
    str: 'string',
    basestring: 'string',
    reqparse.text_type: 'string',
    int: 'integer',
}


def _list_files(path, extension, contains=""):
    """ Returns a list of all the files with the given extension found under the given path. """

    def matches(f):
        return os.path.splitext(f)[1] == '.' + extension and contains in os.path.splitext(f)[0]

    def join_path(dp, f):
        # Remove the static/ prefix. It is added in the template.
        return os.path.join(dp, f)[len('static/'):]

    filepath = os.path.join('static/', path)
    return [join_path(dp, f) for dp, _, files in os.walk(filepath) for f in files if matches(f)]


def render_page_template(name, route_data=None, js_bundle_name=DEFAULT_JS_BUNDLE_NAME, **kwargs):
    """ Renders the page template with the given name as the response and returns its contents. """
    main_scripts = _list_files('build', 'js', js_bundle_name)

    contents = render_template(name,
                               route_data=route_data,
                               main_scripts=main_scripts,
                               **kwargs)

    resp = make_response(contents)
    resp.headers['X-FRAME-OPTIONS'] = 'DENY'
    return resp


def fully_qualified_name(method_view_class):
    return '%s.%s' % (method_view_class.__module__, method_view_class.__name__)


# @lru_cache(maxsize=1)
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()

    print('APP URL MAp:')
    print(app.url_map)
    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}