encode.py

import coreschema
from collections import OrderedDict
from coreapi.compat import urlparse
from openapi_codec.utils import get_method, get_encoding, get_location, get_links_from_document


def generate_swagger_object(document):
    """
    Generates root of the Swagger spec.
    """
    parsed_url = urlparse.urlparse(document.url)

    swagger = OrderedDict()

    swagger['swagger'] = '2.0'
    swagger['info'] = OrderedDict()
    swagger['info']['title'] = document.title
    swagger['info']['description'] = document.description
    swagger['info']['version'] = ''  # Required by the spec

    if parsed_url.netloc:
        swagger['host'] = parsed_url.netloc
    if parsed_url.scheme:
        swagger['schemes'] = [parsed_url.scheme]

    swagger['paths'] = _get_paths_object(document)

    return swagger


def _add_tag_prefix(item):
    operation_id, link, tags = item
    if tags:
        operation_id = tags[0] + '_' + operation_id
    return (operation_id, link, tags)


def _get_links(document):
    """
    Return a list of (operation_id, link, [tags])
    """
    # Extract all the links from the first or second level of the document.
    links = []
    for keys, link in get_links_from_document(document):
        if len(keys) > 1:
            operation_id = '_'.join(keys[1:])
            tags = [keys[0]]
        else:
            operation_id = keys[0]
            tags = []
        links.append((operation_id, link, tags))

    # Determine if the operation ids each have unique names or not.
    operation_ids = [item[0] for item in links]
    unique = len(set(operation_ids)) == len(links)

    # If the operation ids are not unique, then prefix them with the tag.
    if not unique:
        return [_add_tag_prefix(item) for item in links]

    return links


def _get_paths_object(document):
    paths = OrderedDict()

    links = _get_links(document)

    for operation_id, link, tags in links:
        if link.url not in paths:
            paths[link.url] = OrderedDict()

        method = get_method(link)
        operation = _get_operation(operation_id, link, tags)
        paths[link.url].update({method: operation})

    return paths


def _get_operation(operation_id, link, tags):
    encoding = get_encoding(link)
    description = link.description.strip()
    summary = description.splitlines()[0] if description else None

    operation = {
        'operationId': operation_id,
        'responses': _get_responses(link),
        'parameters': _get_parameters(link, encoding)
    }

    if description:
        operation['description'] = description
    if summary:
        operation['summary'] = summary
    if encoding:
        operation['consumes'] = [encoding]
    if tags:
        operation['tags'] = tags
    return operation


def _get_field_description(field):
    if getattr(field, 'description', None) is not None:
        # Deprecated
        return field.description

    if field.schema is None:
        return ''

    return field.schema.description


def _get_field_type(field):
    if getattr(field, 'type', None) is not None:
        # Deprecated
        return field.type

    if field.schema is None:
        return 'string'

    return _get_field_type_name(field.schema.__class__)


def _get_field_type_name(type):
    if type is None:
        return 'string'

    return {
        coreschema.String: 'string',
        coreschema.Integer: 'integer',
        coreschema.Number: 'number',
        coreschema.Boolean: 'boolean',
        coreschema.Array: 'array',
        coreschema.Object: 'object',
    }.get(type, 'string')


def _get_parameters(link, encoding):
    """
    Generates Swagger Parameter Item object.
    """
    parameters = []
    properties = {}
    required = []

    for field in link.fields:
        location = get_location(link, field)
        field_description = _get_field_description(field)
        field_type = _get_field_type(field)
        if location == 'form':
            if encoding in ('multipart/form-data', 'application/x-www-form-urlencoded'):
                # 'formData' in swagger MUST be one of these media types.
                parameter = {
                    'name': field.name,
                    'required': field.required,
                    'in': 'formData',
                    'description': field_description,
                    'type': field_type,
                }
                if field_type == 'array':
                    parameter['items'] = {'type': 'string'}
                parameters.append(parameter)
            else:
                # Expand coreapi fields with location='form' into a single swagger
                # parameter, with a schema containing multiple properties.

                schema_property = {
                    'description': field_description,
                    'type': field_type,
                }
                if field_type == 'array':
                    type_string = 'string'
                    if hasattr(field, "schema") and hasattr(field.schema, "items") and field.schema.items is not None:
                        if isinstance(field.schema.items, type):
                            model_type = field.schema.items
                        else:
                            model_type = field.schema.items.__class__
                        type_string = _get_field_type_name(model_type)
                    schema_property['items'] = {'type': type_string}
                properties[field.name] = schema_property
                if field.required:
                    required.append(field.name)
        elif location == 'body':
            if encoding == 'application/octet-stream':
                # https://github.com/OAI/OpenAPI-Specification/issues/50#issuecomment-112063782
                schema = {'type': 'string', 'format': 'binary'}
            else:
                schema = {}
            parameter = {
                'name': field.name,
                'required': field.required,
                'in': location,
                'description': field_description,
                'schema': schema
            }
            parameters.append(parameter)
        else:
            parameter = {
                'name': field.name,
                'required': field.required,
                'in': location,
                'description': field_description,
                'type': field_type or 'string',
            }
            if field_type == 'array':
                parameter['items'] = {'type': 'string'}
            parameters.append(parameter)

    if properties:
        parameter = {
            'name': 'data',
            'in': 'body',
            'schema': {
                'type': 'object',
                'properties': properties
            }
        }
        if required:
            parameter['schema']['required'] = required
        parameters.append(parameter)

    return parameters


def _get_responses(link):
    """
    Returns minimally acceptable responses object based
    on action / method type.
    """
    template = {'description': ''}
    if link.action.lower() == 'post':
        return {'201': template}
    if link.action.lower() == 'delete':
        return {'204': template}
    return {'200': template}