<commit_before>import json
from contextlib import closing
from urllib2 import urlopen

import yaml
from jsonref import JsonRef
from flask import request
from flask_restful import Api

from . import docs
from .errors import SwaggerError
from .validators import DefaultValidatingDraft4Validator


class SwaggerApi(Api):
    TRUTHY = ('true', 't', 'yes', 'y', 'on', '1')
    FALSY = ('false', 'f', 'no', 'n', 'off', '0')

    def __init__(self, *args, **kwargs):
        self._spec_url = kwargs.pop('spec_url', None)
        self._resource_module = kwargs.pop('resource_module', None)
        self.validate_responses = kwargs.pop('validate_responses', True)
        self.serve_docs = kwargs.pop('serve_docs', '/_docs')
        with closing(urlopen(self._spec_url)) as fp:
            self.spec = yaml.load(fp)
            self._spec = JsonRef.replace_refs(self.spec)
        self._process_spec(self._spec)
        super(SwaggerApi, self).__init__(*args, **kwargs)
        if self.serve_docs is not None:
            self.add_resource(
                docs.Spec,
                '{}/swagger.yaml'.format(self.serve_docs))

    def init_app(self, app):
        super(SwaggerApi, self).init_app(app)

    def add_resource(self, resource, *urls, **kwargs):
        resource_class_args = tuple(kwargs.pop('resource_class_args', ()))
        resource_class_args = (self,) + resource_class_args
        kwargs['resource_class_args'] = resource_class_args
        return super(SwaggerApi, self).add_resource(resource, *urls, **kwargs)

    def _process_spec(self, spec):
        # Catalog the resources handling each path
        self._resource_paths = {}
        if self._resource_module:
            prefix = self._resource_module + '.'
        for path, pspec in spec['paths'].items():
            res = pspec.get('x-resource')
            if res:
                self._resource_paths[prefix + res] = pspec

    def _validate_response(self, resource, response):
        if not self.validate_responses:
            return
        method = request.method.lower()
        resp_spec = self._get_response(resource, method, response.status_code)
        if resp_spec is None:
            raise SwaggerError('Unknown response code {}'.format(
                response.status_code))
        schema_spec = resp_spec.get('schema', None)
        if schema_spec is None:
            return
        schema = DefaultValidatingDraft4Validator(schema_spec)
        json_data = json.loads(response.data)
        schema.validate(json_data)

    def _validate_parameters(self, resource, args, kwargs):
        method = request.method.lower()
        params_spec = self._get_params(resource, method)

        params = {}

        # Check the body param
        body_param_spec = [p for p in params_spec if p['in'] == 'body']
        if body_param_spec:
            data = request.json
            if data is None:
                data = request.form
            params['body'] = self._check_body_param(
                body_param_spec[0], data)

        # Check the primitive params
        params.update(self._check_primitive_params(
            params_spec, 'path', kwargs))
        params.update(self._check_primitive_params(
            params_spec, 'query', request.args))
        params.update(self._check_primitive_params(
            params_spec, 'header', request.headers))
        params.update(self._check_primitive_params(
            params_spec, 'form', request.form))
        return params

    def _get_params(self, resource, method):
        pspec = self._resource_paths.get(resource._resource_name(), {})
        ospec = pspec.get(method, {})
        params = pspec.get('parameters', []) + ospec.get('parameters', [])
        return params

    def _get_response(self, resource, method, status_code):
        pspec = self._resource_paths.get(resource._resource_name(), {})
        ospec = pspec.get(method, {})
        responses = ospec.get('responses', {})
        return responses.get(str(status_code), None)

    def _check_primitive_params(self, params_spec, ptype, data):
        """Validate and convert parameters of a certain primitive type.

        This will verify that *data* correctly validates, and will return
        the validated and converted data.

        Valid ptypes:
         - path
         - query
         - header
         - form
        """
        schema_spec = dict(
            type='object',
            properties={},
            required=[])

        res = dict(data)

        for param in params_spec:
            if param['in'] != ptype:
                continue
            p_name = param['name']
            p_type = param['type']
            p_format = param.get('format', None)
            if param['required']:
                schema_spec['required'].append(p_name)
            schema_spec['properties'][p_name] = p_sch = dict(type=p_type)
            if p_format:
                p_sch['format'] = p_format

            try:
                value = data[p_name]
            except KeyError:
                continue

            # Attempt primitive type conversion
            if p_type == 'integer':
                try:
                    res[p_name] = int(value)
                except:
                    pass
            elif p_type == 'number':
                try:
                    res[p_name] = float(value)
                except:
                    pass
            elif p_type == 'boolean':
                if value.lower() in self.TRUTHY:
                    res[p_name] = True
                elif value.lower() in self.FALSY:
                    res[p_name] = False

        schema = DefaultValidatingDraft4Validator(schema_spec)
        schema.validate(res)
        return res

    def _check_body_param(self, param_spec, data):
        schema = DefaultValidatingDraft4Validator(param_spec['type'])
        schema.validate(data)
        return data<commit_msg>Use Flask config to configure restful-swagger<commit_after>import json
from contextlib import closing
from urllib2 import urlopen

import yaml
from jsonref import JsonRef
from flask import request, jsonify
from flask_restful import Api

from . import docs
from .errors import SwaggerError
from .validators import DefaultValidatingDraft4Validator


class SwaggerApi(Api):
    TRUTHY = ('true', 't', 'yes', 'y', 'on', '1')
    FALSY = ('false', 'f', 'no', 'n', 'off', '0')

    def init_app(self, app):
        self._spec_url = app.config.get(
            'SWAGGER_SPEC_URL', None)
        self._resource_module = app.config.get(
            'SWAGGER_RESOURCE_MODULE', None)
        self.validate_responses = app.config.get(
            'SWAGGER_VALIDATE_RESPONSES', True)
        self.serve_docs = app.config.get(
            'SWAGGER_SERVE_DOCS', '/_docs')
        spec_text = self.get_spec_text()
        self.spec = yaml.load(spec_text)
        self._spec = JsonRef.replace_refs(self.spec)
        self._process_spec(self._spec)

        super(SwaggerApi, self).init_app(app)
        app.register_error_handler(SwaggerError, self.handle_invalid_usage)
        if self.serve_docs:
            app.extensions['flask-restful-swagger'] = self
            app.register_blueprint(docs.mod, url_prefix=self.serve_docs)

    def get_spec_text(self):
        with closing(urlopen(self._spec_url)) as fp:
            return fp.read()

    def add_resource(self, resource, *urls, **kwargs):
        resource_class_args = tuple(kwargs.pop('resource_class_args', ()))
        resource_class_args = (self,) + resource_class_args
        kwargs['resource_class_args'] = resource_class_args
        return super(SwaggerApi, self).add_resource(resource, *urls, **kwargs)

    def handle_invalid_usage(self, error):
        response = jsonify(dict(
            errors=error.errors,
            value=error.value))
        response.status_code = error.status_code
        return response

    def _process_spec(self, spec):
        # Catalog the resources handling each path
        self._resource_paths = {}
        if self._resource_module:
            prefix = self._resource_module + '.'
        for path, pspec in spec['paths'].items():
            res = pspec.get('x-resource')
            if res:
                self._resource_paths[prefix + res] = pspec

    def _validate_response(self, resource, response):
        if not self.validate_responses:
            return
        method = request.method.lower()
        resp_spec = self._get_response(resource, method, response.status_code)
        if resp_spec is None:
            raise SwaggerError(
                500,
                json.loads(response.data),
                'Unknown response code {}'.format(
                    response.status_code))
        schema_spec = resp_spec.get('schema', None)
        if schema_spec is None:
            return
        schema = DefaultValidatingDraft4Validator(schema_spec)
        json_data = json.loads(response.data)
        schema.validate(json_data)

    def _validate_parameters(self, resource, args, kwargs):
        method = request.method.lower()
        params_spec = self._get_params(resource, method)

        params = {}

        # Check the body param
        body_param_spec = [p for p in params_spec if p['in'] == 'body']
        if body_param_spec:
            data = request.json
            if data is None:
                data = request.form
            params['body'] = self._check_body_param(
                body_param_spec[0], data)

        # Check the primitive params
        params.update(self._check_primitive_params(
            params_spec, 'path', kwargs))
        params.update(self._check_primitive_params(
            params_spec, 'query', request.args))
        params.update(self._check_primitive_params(
            params_spec, 'header', request.headers))
        params.update(self._check_primitive_params(
            params_spec, 'form', request.form))
        return params

    def _get_params(self, resource, method):
        pspec = self._resource_paths.get(resource._resource_name(), {})
        ospec = pspec.get(method, {})
        params = pspec.get('parameters', []) + ospec.get('parameters', [])
        return params

    def _get_response(self, resource, method, status_code):
        pspec = self._resource_paths.get(resource._resource_name(), {})
        ospec = pspec.get(method, {})
        responses = ospec.get('responses', {})
        return responses.get(str(status_code), None)

    def _check_primitive_params(self, params_spec, ptype, data):
        """Validate and convert parameters of a certain primitive type.

        This will verify that *data* correctly validates, and will return
        the validated and converted data.

        Valid ptypes:
         - path
         - query
         - header
         - form
        """
        schema_spec = dict(
            type='object',
            properties={},
            required=[])

        res = dict(data)

        for param in params_spec:
            if param['in'] != ptype:
                continue
            p_name = param['name']
            p_type = param['type']
            p_format = param.get('format', None)
            if param['required']:
                schema_spec['required'].append(p_name)
            schema_spec['properties'][p_name] = p_sch = dict(type=p_type)
            if p_format:
                p_sch['format'] = p_format
            if 'default' in param:
                p_sch['default'] = param['default']

            try:
                value = data[p_name]
            except KeyError:
                continue

            # Attempt primitive type conversion
            if p_type == 'integer':
                try:
                    res[p_name] = int(value)
                except:
                    pass
            elif p_type == 'number':
                try:
                    res[p_name] = float(value)
                except:
                    pass
            elif p_type == 'boolean':
                if value.lower() in self.TRUTHY:
                    res[p_name] = True
                elif value.lower() in self.FALSY:
                    res[p_name] = False

        schema = DefaultValidatingDraft4Validator(schema_spec)
        schema.validate(res)
        return res

    def _check_body_param(self, param_spec, data):
        schema = DefaultValidatingDraft4Validator(param_spec['schema'])
        schema.validate(data)
        return data
{'fraction_non_alphanumeric': 0.10214751065712217, 'fraction_numerical': 0.0015281911043191507, 'mean_word_length': 2.067850974586726, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2201338', 'n_tokens_mistral': 3501, 'n_tokens_neox': 3418, 'n_words': 790}