Re-implementing the Multi-part Form Parser in APIStar >= 0.4

apistar_forms.py

import io
import typing
from apistar import Component, http
from apistar.server.wsgi import WSGIEnviron
from werkzeug.datastructures import ImmutableMultiDict
from werkzeug.formparser import FormDataParser
from werkzeug.http import parse_options_header
from werkzeug.wsgi import get_input_stream

RequestStream = typing.NewType('RequestStream', io.BufferedIOBase)
MultiPartForm = typing.NewType('MultiPartForm', ImmutableMultiDict)


class RequestStreamComponent(Component):
    def resolve(self, environ: WSGIEnviron) -> RequestStream:
        return get_input_stream(environ)


class MultiPartParser(Component):
    media_type = 'multipart/form-data'

    def get_content_length(self, headers: http.Headers) -> typing.Optional[int]:
        content_length = headers.get('Content-Length')
        if content_length is not None:
            try:
                return max(0, int(content_length))
            except (ValueError, TypeError):
                pass
        return None

    def get_mimetype_and_options(self, headers: http.Headers) -> typing.Tuple[str, dict]:
        content_type = headers.get('Content-Type')
        if content_type:
            return parse_options_header(content_type)
        return '', {}

    def resolve(self, headers: http.Headers, stream: RequestStream) -> MultiPartForm:
        mimetype, options = self.get_mimetype_and_options(headers)
        content_length = self.get_content_length(headers)
        parser = FormDataParser()
        stream, form, files = parser.parse(stream, mimetype, content_length, options)
        return ImmutableMultiDict(list(form.items()) + list(files.items()))