Instantly share code, notes, and snippets.

Embed
What would you like to do?
Re-implementing the Multi-part Form Parser in APIStar >= 0.4
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()))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment