Repository: falconry/falcon
Pull request #1590 (2019-10-25 19:01:04)

feat(media): implement multipart form media handler
@kgriffs @jmvrbanac @nZac and others in the community who are following the development of this feature:
this is still rough on edges when it comes to documentation, but the interface and the implementation are hopefully alpha-level ready.

I have also created an extended test/benchmark suite for tuning performance as well as doubling as extra validation for code correctness: https://github.com/vytas7/falcon-multipart-tests

There is still one notable case of many small reads from the part stream which I stopped short of optimizing. The machinery is largely ready, I could just force reads to be aligned on the part stream. But I would be happy to get feedback from the community how the functionality is used in the wild in order to get a better impression whether we need to focus on this specific usage pattern.

**The Gist** illustrating the interface and the demo therein are still valid: https://gist.github.com/vytas7/34c60e5ac3a4bc2f2eb0af2428d77003 (I have actually just slightly updated it).

I have also started implementing **async multipart form parsing for ASGI** (i.e. marrying this to async media support in https://github.com/falconry/falcon/pull/1573).
This class could probably be used verbatim for ASGI from #1573, but better don't even try it at home. In the best case, your file would be buffered in memory in a bytes buffer.


_NB: creating a new pull request because my previous branch became too complex for GitHub to handle (the GH PR interface was warning of bogus conflicts)._


vytas7: Implements #953
codecov[bot]: # [Codecov](https://codecov.io/gh/falconry/falcon/pull/1590?src=pr&el=h1) Report
> Merging [#1590](https://codecov.io/gh/falconry/falcon/pull/1590?src=pr&el=desc) into [master](https://codecov.io/gh/falconry/falcon/commit/1e7e8c0a43debb9f56cb7b53cd01a89002e9e80d?src=pr&el=desc) will **not change** coverage.
> The diff coverage is `100%`.

[![Impacted file tree graph](https://codecov.io/gh/falconry/falcon/pull/1590/graphs/tree.svg?width=650&token=zOtkfNQ3h6&height=150&src=pr)](https://codecov.io/gh/falconry/falcon/pull/1590?src=pr&el=tree)

```diff
@@           Coverage Diff           @@
##           master   #1590    +/-   ##
=======================================
  Coverage     100%    100%            
=======================================
  Files          39      41     +2     
  Lines        2578    2904   +326     
  Branches      393     453    +60     
=======================================
+ Hits         2578    2904   +326
```


| [Impacted Files](https://codecov.io/gh/falconry/falcon/pull/1590?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [falcon/request.py](https://codecov.io/gh/falconry/falcon/pull/1590/diff?src=pr&el=tree#diff-ZmFsY29uL3JlcXVlc3QucHk=) | `100% <100%> (ø)` | :arrow_up: |
| [falcon/util/streams.py](https://codecov.io/gh/falconry/falcon/pull/1590/diff?src=pr&el=tree#diff-ZmFsY29uL3V0aWwvc3RyZWFtcy5weQ==) | `100% <100%> (ø)` | |
| [falcon/util/\_\_init\_\_.py](https://codecov.io/gh/falconry/falcon/pull/1590/diff?src=pr&el=tree#diff-ZmFsY29uL3V0aWwvX19pbml0X18ucHk=) | `100% <100%> (ø)` | :arrow_up: |
| [falcon/constants.py](https://codecov.io/gh/falconry/falcon/pull/1590/diff?src=pr&el=tree#diff-ZmFsY29uL2NvbnN0YW50cy5weQ==) | `100% <100%> (ø)` | :arrow_up: |
| [falcon/util/misc.py](https://codecov.io/gh/falconry/falcon/pull/1590/diff?src=pr&el=tree#diff-ZmFsY29uL3V0aWwvbWlzYy5weQ==) | `100% <100%> (ø)` | :arrow_up: |
| [falcon/media/\_\_init\_\_.py](https://codecov.io/gh/falconry/falcon/pull/1590/diff?src=pr&el=tree#diff-ZmFsY29uL21lZGlhL19faW5pdF9fLnB5) | `100% <100%> (ø)` | :arrow_up: |
| [falcon/media/base.py](https://codecov.io/gh/falconry/falcon/pull/1590/diff?src=pr&el=tree#diff-ZmFsY29uL21lZGlhL2Jhc2UucHk=) | `100% <100%> (ø)` | :arrow_up: |
| [falcon/media/multipart.py](https://codecov.io/gh/falconry/falcon/pull/1590/diff?src=pr&el=tree#diff-ZmFsY29uL21lZGlhL211bHRpcGFydC5weQ==) | `100% <100%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/falconry/falcon/pull/1590?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/falconry/falcon/pull/1590?src=pr&el=footer). Last update [1e7e8c0...9e9ab65](https://codecov.io/gh/falconry/falcon/pull/1590?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

vytas7: > I am still slowly walking through this. Here are some very minor pieces of feedback until I can give this a more thorough look.
> 
> In the meantime, can you give me a synopsis on the addition of `BufferedStream`? Is it all the handle the `boundry` stuff?

@nZac Thanks for looking at this PR! Yeah, `BufferedStream` is to handle the boundary stuff. Once you read a chunk from the WSGI input stream, and find the part boundary, you need to buffer the remaining part to be able to transparently present it as the part (sub-)stream.

The ASGI variant has a bit simpler variant of dealing with the same problem (the ASGI HTTP request input is coming in the form of chunked messages).
We'll probably have to sync approaches with @kgriffs (see `falcon.asgi.stream`). Also, we may want to retire the existing BoundedStream in favour of this BufferedStream (or w/e name we choose for it). See also this issue: 
kgriffs: > IIRC I probably did it this way because the part stream's read_until() is exposed to API users, so the generic MultipartParseError would sound a bit off. But we could maybe define a new error type for read_until(), and re-raise that in the multipart handler?

That could work. BufferedStream can define its own exception types and raise those as needed. 
vytas7: @kgriffs @nZac @jmvrbanac I'm still working on hardening the unit test suite, but I don't think the review should block on that. Ready to go for the second round!!!

Furthermore, the default `BufferedReader` chunk size is still TBD. Opinions are very welcome!
Some data to assist in choosing: https://travis-ci.org/vytas7/falcon-multipart-tests/jobs/644933725
I'm a bit divided between 32 KiB (currently presented in this PR) and 64 KiB.
64 KiB seems to give the best performance, however, with Cython helping to do all that pesky offset match virtually "for free" (compared to Python statements), the difference is not huge between the two. And I'm not a fan of buffer bloat.
kgriffs: > I'm a bit divided between 32 KiB (currently presented in this PR) and 64 KiB.

32K seems like a good default in terms of balancing speed and memory usage. I say stick with that but also document that the app developer may want to change `DEFAULT_CHUNK_SIZE` (or perhaps this should be a `RequestOptions` field?) to 64K if they frequently read large file uploads and have enough RAM capacity to handle it.

Also, should we be worried about any potentially nasty surprises if the value of `io.DEFAULT_BUFFER_SIZE` ever changes?
vytas7: > 32K seems like a good default in terms of balancing speed and memory usage. I say stick with that but also document that the app developer may want to change `DEFAULT_CHUNK_SIZE` (or perhaps this should be a `RequestOptions` field?) to 64K if they frequently read large parts (file uploads) and have enough RAM capacity to handle it.

:+1: Sounds good.
I've stashed this changed locally so far though; I need to take a look again with a fresh head and see what's the best way to make `RequestOptions` accessible from media handlers. Atm the `req` object is not passed down to these.

> Also, should we be worried about any potentially nasty surprises if the value of `io.DEFAULT_BUFFER_SIZE` ever changes?

Good point. I changed it to an explicit number.

Patch:
diff --git a/docs/_newsfragments/multipart-form-media-handler.feature.rst b/docs/_newsfragments/multipart-form-media-handler.feature.rst
new file mode 100644
index 0000000..8c57437
--- /dev/null
+++ b/docs/_newsfragments/multipart-form-media-handler.feature.rst
@@ -0,0 +1,2 @@
+``MultipartFormHandler`` was added for handling multipart forms (of content
+type ``multipart/form-data``) as ``Request.media``.
diff --git a/docs/api/index.rst b/docs/api/index.rst
index 8aa3c0b..09ce36d 100644
--- a/docs/api/index.rst
+++ b/docs/api/index.rst
@@ -10,6 +10,7 @@ Framework Reference
    status
    errors
    media
+   multipart
    redirects
    middleware
    cors
diff --git a/docs/api/media.rst b/docs/api/media.rst
index 9041ea3..97b6c4d 100644
--- a/docs/api/media.rst
+++ b/docs/api/media.rst
@@ -4,8 +4,8 @@ Media
 =====
 
 Falcon allows for easy and customizable internet media type handling. By
-default Falcon only enables handlers for JSON and URL-encoded HTML
-forms. However, additional handlers can be configured through the
+default Falcon only enables handlers for JSON and HTML (URL-encoded and
+multipart) forms. However, additional handlers can be configured through the
 :any:`falcon.RequestOptions` and :any:`falcon.ResponseOptions` objects
 specified on your :any:`falcon.App`.
 
@@ -123,6 +123,9 @@ Supported Handler Types
 .. autoclass:: falcon.media.MessagePackHandler
     :members:
 
+.. autoclass:: falcon.media.MultipartFormHandler
+    :members:
+
 .. autoclass:: falcon.media.URLEncodedFormHandler
     :members:
 
@@ -157,6 +160,7 @@ common media types, including the following:
 
     falcon.MEDIA_JSON
     falcon.MEDIA_MSGPACK
+    falcon.MEDIA_MULTIPART
     falcon.MEDIA_URLENCODED
     falcon.MEDIA_YAML
     falcon.MEDIA_XML
diff --git a/docs/api/multipart.rst b/docs/api/multipart.rst
new file mode 100644
index 0000000..b37db2f
--- /dev/null
+++ b/docs/api/multipart.rst
@@ -0,0 +1,56 @@
+.. _multipart:
+
+Multipart Forms
+===============
+
+Falcon features easy and efficient access to submitted multipart forms by using
+:class:`falcon.media.MultipartFormHandler` to handle the
+``multipart/form-data`` :ref:`media <media>` type. This handler is enabled by
+default, allowing you to use ``req.media`` to iterate over the
+:class:`body parts <falcon.media.multipart.BodyPart>` in a form:
+
+.. code:: python
+
+    for part in req.media:
+        if part.content_type == 'application/json':
+            # TODO: Body part is a JSON document, do something useful with it
+            resp.media = part.media
+        elif part.name == 'datafile':
+            while True:
+                # Do something with the uploaded data (file)
+                chunk = part.stream.read(8192)
+                if not chunk:
+                    break
+                feed_data(chunk)
+        elif part.name == 'imagedata':
+            # Store this body part in a file.
+            with open(path.join(UPLOADS, part.secure_filename), 'wb') as dest:
+                part.stream.pipe(dest)
+        else:
+            # TODO: Do something else
+
+.. note::
+   The request stream is only consumed along iteration over the body parts in a
+   form, rather than being read in and buffered all at once.
+
+   You can then choose whether to read the whole part into the memory, store in
+   a file, or :ref:`upload to the cloud <multipart_cloud_upload>`.
+   Falcon offers straightforward support for all these scenarios.
+
+Body Part Type
+--------------
+
+.. autoclass:: falcon.media.multipart.BodyPart
+    :members:
+
+Parsing Options
+---------------
+
+.. autoclass:: falcon.media.multipart.MultipartParseOptions
+    :members:
+
+Parsing Errors
+--------------
+
+.. autoclass:: falcon.media.multipart.MultipartParseError
+    :members:
diff --git a/docs/api/util.rst b/docs/api/util.rst
index 088f19a..8533c0a 100644
--- a/docs/api/util.rst
+++ b/docs/api/util.rst
@@ -15,7 +15,8 @@ Miscellaneous
 
 .. automodule:: falcon
     :members: deprecated, http_now, dt_to_http, http_date_to_dt,
-        to_query_str, get_http_status, get_bound_method, is_python_func
+        to_query_str, get_http_status, get_bound_method, secure_filename,
+        is_python_func
 
 .. autoclass:: falcon.TimezoneGMT
     :members:
diff --git a/docs/user/faq.rst b/docs/user/faq.rst
index b05f842..42e4df4 100644
--- a/docs/user/faq.rst
+++ b/docs/user/faq.rst
@@ -562,41 +562,53 @@ POSTed form parameters may also be read directly from
 :meth:`falcon.uri.parse_query_string` or
 `urllib.parse.parse_qs() <https://docs.python.org/3.6/library/urllib.parse.html#urllib.parse.parse_qs>`_.
 
+.. _access_multipart_files:
+
 How can I access POSTed files?
 ------------------------------
-Falcon does not currently support parsing files submitted by
-an HTTP form (``multipart/form-data``), although we do plan
-to add this feature in a future version. In the meantime,
-you can use the standard ``cgi.FieldStorage`` class to
-parse the request:
+
+If files are ``POST``\ed as part of a :ref:`multipart form <multipart>`, the
+default :class:`MultipartFormHandler <falcon.media.MultipartFormHandler>` can
+be used to efficiently parse the submitted ``multipart/form-data``
+:ref:`request media <media>` by iterating over the multipart
+:class:`body parts <falcon.media.multipart.BodyPart>`:
 
 .. code:: python
 
-    # TODO: Either validate that content type is multipart/form-data
-    # here, or in another hook before allowing execution to proceed.
+    for part in req.media:
+        # TODO: Do something with the body part
+        pass
+
+.. _multipart_cloud_upload:
 
-    # This must be done to avoid a bug in cgi.FieldStorage
-    env = req.env
-    env.setdefault('QUERY_STRING', '')
+How can I save POSTed files (from a multipart form) directly to AWS S3?
+-----------------------------------------------------------------------
 
-    # TODO: Add error handling, when the request is not formatted
-    # correctly or does not contain the desired field...
+As highlighted in the previous answer dealing with
+:ref:`files posted as multipart form <access_multipart_files>`,
+:class:`falcon.media.MultipartFormHandler` may be used to iterate over the
+uploaded multipart body parts.
 
-    # TODO: Consider overriding make_file, so that you can
-    # stream directly to the destination rather than
-    # buffering using TemporaryFile (see http://goo.gl/Yo8h3P)
-    form = cgi.FieldStorage(fp=req.stream, environ=env)
+The `stream` of a body part is a file-like object implementing the ``read()``
+method, making it compatible with ``boto3``\'s
+`upload_fileobj <https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#S3.Client.upload_fileobj>`_:
 
-    file_item = form[name]
-    if file_item.file:
-        # TODO: It's an uploaded file... read it in
-    else:
-        # TODO: Raise an error
+.. code:: python
 
-You might also try this
-`streaming_form_data <https://streaming-form-data.readthedocs.io/en/latest/>`_
-package by Siddhant Goel, or searching PyPI for additional options from the
-community.
+    import boto3
+    s3 = boto3.client('s3')
+
+    # ...
+
+    for part in req.media:
+        if part.name == 'myfile':
+            s3.upload_fileobj(part.stream, 'mybucket', 'mykey')
+
+.. note::
+   Falcon is not endorsing any particular cloud service provider, and AWS S3
+   and ``boto3`` are referenced here just as a popular example. The same
+   pattern can be applied to any storage API that supports streaming directly
+   from a file-like object.
 
 How do I consume a query string that has a JSON value?
 ------------------------------------------------------
diff --git a/falcon/constants.py b/falcon/constants.py
index 420de21..40e39da 100644
--- a/falcon/constants.py
+++ b/falcon/constants.py
@@ -51,6 +51,8 @@ MEDIA_JSON = 'application/json'
 # but the use of the 'x-' prefix is discouraged by RFC 6838.
 MEDIA_MSGPACK = 'application/msgpack'
 
+MEDIA_MULTIPART = 'multipart/form-data'
+
 MEDIA_URLENCODED = 'application/x-www-form-urlencoded'
 
 # NOTE(kgriffs): An internet media type for YAML has not been
diff --git a/falcon/cyutil/reader.pyx b/falcon/cyutil/reader.pyx
new file mode 100644
index 0000000..46be840
--- /dev/null
+++ b/falcon/cyutil/reader.pyx
@@ -0,0 +1,416 @@
+# Copyright 2019-2020 by Vytautas Liuolia.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Buffered stream reader (cythonized variant)."""
+
+from libc.stdint cimport uint32_t
+
+import functools
+import io
+
+from falcon.util.reader import DEFAULT_CHUNK_SIZE, DelimiterError, _MAX_JOIN_CHUNKS
+
+
+cdef class BufferedReader:
+
+    cdef _read_func
+    cdef Py_ssize_t _chunk_size
+    cdef Py_ssize_t _max_join_size
+
+    cdef bytes _buffer
+    cdef Py_ssize_t _buffer_len
+    cdef Py_ssize_t _buffer_pos
+    cdef Py_ssize_t _max_bytes_remaining
+
+    def __cinit__(self, read, max_stream_len, chunk_size=None):
+        self._read_func = read
+        self._chunk_size = chunk_size or DEFAULT_CHUNK_SIZE
+        self._max_join_size = self._chunk_size * _MAX_JOIN_CHUNKS
+
+        self._buffer = b''
+        self._buffer_len = 0
+        self._buffer_pos = 0
+        self._max_bytes_remaining = max_stream_len
+
+    cdef bytes _perform_read(self, Py_ssize_t size):
+        cdef bytes chunk
+        cdef Py_ssize_t chunk_len
+        cdef result
+
+        size = min(size, self._max_bytes_remaining)
+        if size <= 0:
+            return b''
+
+        chunk = self._read_func(size)
+        chunk_len = len(chunk)
+        self._max_bytes_remaining -= chunk_len
+        if chunk_len == size:
+            return chunk
+
+        if chunk_len == 0:
+            # NOTE(vytas): The EOF.
+            self._max_bytes_remaining = 0
+            return b''
+
+        result = io.BytesIO(chunk)
+        result.seek(chunk_len)
+
+        while True:
+            size -= chunk_len
+            if size <= 0:
+                return result.getvalue()
+
+            chunk = self._read_func(size)
+            chunk_len = len(chunk)
+            if chunk_len == 0:
+                # NOTE(vytas): The EOF.
+                self._max_bytes_remaining = 0
+                return result.getvalue()
+
+            self._max_bytes_remaining -= chunk_len
+            result.write(chunk)
+
+    cdef _fill_buffer(self):
+        cdef Py_ssize_t read_size
+
+        if self._buffer_len - self._buffer_pos < self._chunk_size:
+            read_size = self._chunk_size - (self._buffer_len - self._buffer_pos)
+
+            if self._buffer_pos == 0:
+                self._buffer += self._perform_read(read_size)
+            else:
+                self._buffer = (self._buffer[self._buffer_pos:] +
+                                self._perform_read(read_size))
+                self._buffer_pos = 0
+
+            self._buffer_len = len(self._buffer)
+
+    def peek(self, Py_ssize_t size=-1):
+        if size < 0 or size > self._chunk_size:
+            size = self._chunk_size
+
+        if self._buffer_len - self._buffer_pos < size:
+            self._fill_buffer()
+
+        return self._buffer[self._buffer_pos:self._buffer_pos + size]
+
+    cdef Py_ssize_t _normalize_size(self, size):
+        cdef Py_ssize_t result
+        cdef Py_ssize_t max_size = (self._max_bytes_remaining + self._buffer_len -
+                                    self._buffer_pos)
+
+        if size is None:
+            return max_size
+
+        # PERF(vytas): Start operating on a Py_ssize_t as soon as the None case
+        #   is ruled out.
+        result = size
+
+        if result == -1 or result > max_size:
+            return max_size
+        return result
+
+    def read(self, size=-1):
+        return self._read(self._normalize_size(size))
+
+    cdef _read(self, Py_ssize_t size):
+        cdef Py_ssize_t read_size
+        cdef bytes result
+
+        # NOTE(vytas): Dish directly from the buffer, if possible.
+        if size <= self._buffer_len - self._buffer_pos:
+            if size == self._buffer_len and self._buffer_pos == 0:
+                result = self._buffer
+                self._buffer_len = 0
+                self._buffer = b''
+                return result
+
+            self._buffer_pos += size
+            return self._buffer[self._buffer_pos - size:self._buffer_pos]
+
+        # NOTE(vytas): Pass through large reads.
+        if self._buffer_len == 0 and size >= self._chunk_size:
+            return self._perform_read(size)
+
+        # NOTE(vytas): if size > self._buffer_len - self._buffer_pos
+        read_size = size - (self._buffer_len - self._buffer_pos)
+        result = self._buffer[self._buffer_pos:]
+
+        if read_size >= self._chunk_size:
+            self._buffer_len = 0
+            self._buffer_pos = 0
+            self._buffer = b''
+            return result + self._perform_read(read_size)
+
+        self._buffer = self._perform_read(self._chunk_size)
+        self._buffer_len = len(self._buffer)
+        self._buffer_pos = read_size
+        return result + self._buffer[:read_size]
+
+    def read_until(self, bytes delimiter not None, size=-1,
+                   consume_delimiter=False):
+        cdef Py_ssize_t read_size = self._normalize_size(size)
+        cdef result
+
+        read_size = self._normalize_size(size)
+        if read_size <= self._max_join_size:
+            return self._read_until(delimiter, read_size, consume_delimiter)
+
+        # NOTE(vytas): A large size was requested, optimize for memory
+        #   consumption by avoiding to momentarily keep both the chunks and the
+        #   joint result in memory at the same time.
+        result = io.BytesIO()
+        self.pipe_until(delimiter, result, consume_delimiter, read_size)
+        return result.getvalue()
+
+    cdef _finalize_read_until(
+        self, Py_ssize_t size, backlog, Py_ssize_t have_bytes,
+        Py_ssize_t consume_bytes, bytes delimiter=None,
+        Py_ssize_t delimiter_pos=-1,
+        bytes next_chunk=None, Py_ssize_t next_chunk_len=0):
+
+        if delimiter_pos < 0 and delimiter is not None:
+            delimiter_pos = self._buffer.find(delimiter, self._buffer_pos)
+
+        if delimiter_pos >= 0:
+            size = min(size, have_bytes + delimiter_pos - self._buffer_pos)
+
+        if have_bytes == 0:
+            # PERF(vytas): Do not join bytes unless needed.
+            ret_value = self._read(size)
+        else:
+            backlog.append(self._read(size - have_bytes))
+            ret_value = b''.join(backlog)
+
+        if next_chunk_len > 0:
+            if self._buffer_len == 0:
+                self._buffer = next_chunk
+                self._buffer_len = next_chunk_len
+            else:
+                self._buffer = self._buffer[self._buffer_pos:] + next_chunk
+                self._buffer_len = (self._buffer_len - self._buffer_pos +
+                                    next_chunk_len)
+                self._buffer_pos = 0
+
+        if consume_bytes:
+            if delimiter_pos < 0:
+                if self.peek(consume_bytes) != delimiter:
+                    raise DelimiterError('expected delimiter missing')
+            elif self._buffer_pos != delimiter_pos:
+                # NOTE(vytas): If we are going to consume the delimiter the
+                #   quick way (i.e., skipping the above peek() check), we must
+                #   make sure it is directly succeeding the result.
+                raise DelimiterError('expected delimiter missing')
+
+            self._buffer_pos += consume_bytes
+
+        return ret_value
+
+    cdef _read_until(self, bytes delimiter, Py_ssize_t size,
+                     bint consume_delimiter):
+        cdef list result = []
+        cdef Py_ssize_t have_bytes = 0
+        cdef Py_ssize_t delimiter_len_1 = len(delimiter) - 1
+        cdef Py_ssize_t delimiter_pos = -1
+        cdef Py_ssize_t consume_bytes = (delimiter_len_1 + 1) if consume_delimiter else 0
+        cdef Py_ssize_t offset
+
+        if not 0 <= delimiter_len_1 < self._chunk_size:
+            raise ValueError('delimiter length must be within [1, chunk_size]')
+
+        # PERF(vytas): If the requested size is equal to the chunk size (or is
+        #   a multiple of it), align the buffer.
+        #   This can often nearly *double* the performance if one is reading in
+        #   chunks.
+        if size % self._chunk_size == 0:
+            self._fill_buffer()
+
+        # PERF(vytas): Quickly check for the first 4 delimiter bytes.
+        cdef bint quick_found = True
+        cdef uint32_t delimiter_head = 0
+        cdef uint32_t current
+        cdef Py_ssize_t index
+        cdef const unsigned char* ptr
+        if 3 <= delimiter_len_1 < 128:
+            ptr = delimiter
+            delimiter_head = (
+                (ptr[0] << 24) | (ptr[1] << 16) | (ptr[2] << 8) | ptr[3])
+
+        while True:
+            if self._buffer_len > self._buffer_pos:
+                delimiter_pos = self._buffer.find(delimiter, self._buffer_pos)
+                if delimiter_pos >= 0:
+                    # NOTE(vytas): Delimiter was found in the current buffer.
+                    #   We can now return to the caller.
+                    return self._finalize_read_until(
+                        size, result, have_bytes, consume_bytes, None,
+                        delimiter_pos)
+
+            if size < (have_bytes + self._buffer_len - self._buffer_pos -
+                       delimiter_len_1):
+                # NOTE(vytas): We now have enough data in the buffer to return
+                #   to the caller.
+                return self._finalize_read_until(
+                    size, result, have_bytes, consume_bytes, delimiter)
+
+            next_chunk = self._perform_read(self._chunk_size)
+            next_chunk_len = len(next_chunk)
+            if self._max_bytes_remaining == 0:
+                # NOTE(vytas): We have reached the EOF.
+                self._buffer_len += next_chunk_len
+                self._buffer += next_chunk
+                return self._finalize_read_until(
+                    size, result, have_bytes, consume_bytes, delimiter)
+
+            # NOTE(vytas): The buffer was empty before, skip straight to the
+            #   next chunk.
+            if self._buffer_len <= self._buffer_pos:
+                self._buffer_len = next_chunk_len
+                self._buffer_pos = 0
+                self._buffer = next_chunk
+                continue
+
+            # NOTE(vytas): We must check there is no delimiter in the chunk
+            #   boundary before we can safely splice them.
+            if delimiter_len_1 > 0:
+                offset = max(self._buffer_len - delimiter_len_1,
+                             self._buffer_pos)
+
+                # PERF(vytas): Quickly check for the first 4 delimiter bytes.
+                #   This is pure Cython code with no counterpart in streams.py.
+                if 3 <= delimiter_len_1 < 128:
+                    quick_found = False
+                    ptr = self._buffer
+                    current = 0
+                    for index in range(offset, self._buffer_len):
+                        current <<= 8
+                        current |= ptr[index]
+                        if current == delimiter_head:
+                            quick_found = True
+                            break
+
+                    if not quick_found:
+                        ptr = next_chunk
+                        for index in range(3):
+                            current <<= 8
+                            current |= ptr[index]
+                            if current == delimiter_head:
+                                quick_found = True
+                                break
+
+                if quick_found:
+                    offset = max(self._buffer_len - delimiter_len_1,
+                                 self._buffer_pos)
+                    fragment = (self._buffer[offset:] +
+                                next_chunk[:delimiter_len_1])
+                    delimiter_pos = fragment.find(delimiter)
+                    if delimiter_pos >= 0:
+                        self._buffer_len += next_chunk_len
+                        self._buffer += next_chunk
+                        return self._finalize_read_until(
+                            size, result, have_bytes, consume_bytes, delimiter,
+                            delimiter_pos + offset)
+
+            if have_bytes + self._buffer_len - self._buffer_pos >= size:
+                # NOTE(vytas): we have now verified that all bytes currently in
+                #   the buffer are delimiter-free, including the border of the
+                #   upcoming chunk
+                return self._finalize_read_until(
+                    size, result, have_bytes, consume_bytes, delimiter, -1,
+                    next_chunk, next_chunk_len)
+
+            have_bytes += self._buffer_len - self._buffer_pos
+            if self._buffer_pos > 0:
+                result.append(self._buffer[self._buffer_pos:])
+            else:
+                result.append(self._buffer)
+            self._buffer_len = next_chunk_len
+            self._buffer_pos = 0
+            self._buffer = next_chunk
+
+    def pipe(self, destination=None):
+        while True:
+            chunk = self.read(self._chunk_size)
+            if not chunk:
+                break
+
+            if destination is not None:
+                destination.write(chunk)
+
+    def pipe_until(self, delimiter, destination=None, consume_delimiter=False,
+                   _size=None):
+        cdef Py_ssize_t remaining = self._normalize_size(_size)
+
+        while remaining > 0:
+            chunk = self._read_until(
+                delimiter, min(self._chunk_size, remaining), False)
+            if not chunk:
+                break
+
+            if destination is not None:
+                destination.write(chunk)
+
+            remaining -= self._chunk_size
+
+        if consume_delimiter:
+            delimiter_len = len(delimiter)
+            if self.peek(delimiter_len) != delimiter:
+                raise DelimiterError('expected delimiter missing')
+            self._buffer_pos += delimiter_len
+
+    def exhaust(self):
+        self.pipe()
+
+    def delimit(self, delimiter):
+        read = functools.partial(self.read_until, delimiter)
+        return type(self)(read, self._normalize_size(None), self._chunk_size)
+
+    def readline(self, size=-1):
+        size = self._normalize_size(size)
+
+        result = self.read_until(b'\n', size)
+        if len(result) < size:
+            return result + self.read(1)
+        return result
+
+    def readlines(self, hint=-1):
+        cdef Py_ssize_t read = 0
+        cdef list result = []
+
+        while True:
+            line = self.readline()
+            if not line:
+                break
+            result.append(line)
+
+            if hint >= 0:
+                read += len(line)
+                if read >= hint:
+                    break
+
+        return result
+
+    # --- implementing IOBase methods, the duck-typing way ---
+
+    def readable(self):
+        """Always returns ``True``."""
+        return True
+
+    def seekable(self):
+        """Always returns ``False``."""
+        return False
+
+    def writeable(self):
+        """Always returns ``False``."""
+        return False
diff --git a/falcon/media/__init__.py b/falcon/media/__init__.py
index c598c8f..0e1d156 100644
--- a/falcon/media/__init__.py
+++ b/falcon/media/__init__.py
@@ -1,6 +1,16 @@
-from .base import BaseHandler  # NOQA
-from .json import JSONHandler  # NOQA
-from .msgpack import MessagePackHandler  # NOQA
-from .urlencoded import URLEncodedFormHandler  # NOQA
+from .base import BaseHandler
+from .handlers import Handlers
+from .json import JSONHandler
+from .msgpack import MessagePackHandler
+from .multipart import MultipartFormHandler
+from .urlencoded import URLEncodedFormHandler
 
-from .handlers import Handlers  # NOQA
+
+__all__ = [
+    'BaseHandler',
+    'Handlers',
+    'JSONHandler',
+    'MessagePackHandler',
+    'MultipartFormHandler',
+    'URLEncodedFormHandler',
+]
diff --git a/falcon/media/base.py b/falcon/media/base.py
index 55b34a5..a96c8a9 100644
--- a/falcon/media/base.py
+++ b/falcon/media/base.py
@@ -97,7 +97,6 @@ class BaseHandler(metaclass=abc.ABCMeta):
         Returns:
             object: A deserialized object.
         """
-
         data = await stream.read()
 
         # NOTE(kgriffs): Override content length to make sure it is correct,
@@ -105,3 +104,11 @@ class BaseHandler(metaclass=abc.ABCMeta):
         content_length = len(data)
 
         return self.deserialize(io.BytesIO(data), content_type, content_length)
+
+    exhaust_stream = False
+    """Whether to exhaust the WSGI input stream upon finishing deserialization.
+
+    Exhausting the stream may be useful for handlers that do not necessarily
+    consume the whole stream, but the deserialized media object is complete and
+    does not involve further streaming.
+    """
diff --git a/falcon/media/handlers.py b/falcon/media/handlers.py
index 1a083da..62eea20 100644
--- a/falcon/media/handlers.py
+++ b/falcon/media/handlers.py
@@ -1,9 +1,10 @@
 from collections import UserDict
 
 from falcon import errors
-from falcon.constants import MEDIA_URLENCODED
-from falcon.media import JSONHandler
-from falcon.media import URLEncodedFormHandler
+from falcon.constants import MEDIA_MULTIPART, MEDIA_URLENCODED
+from falcon.media.json import JSONHandler
+from falcon.media.multipart import MultipartFormHandler, MultipartParseOptions
+from falcon.media.urlencoded import URLEncodedFormHandler
 from falcon.vendor import mimeparse
 
 
@@ -13,6 +14,7 @@ class Handlers(UserDict):
         handlers = initial or {
             'application/json': JSONHandler(),
             'application/json; charset=UTF-8': JSONHandler(),
+            MEDIA_MULTIPART: MultipartFormHandler(),
             MEDIA_URLENCODED: URLEncodedFormHandler(),
         }
 
@@ -54,3 +56,11 @@ class Handlers(UserDict):
             )
 
         return self.data[resolved]
+
+
+# NOTE(vytas): An ugly way to work around circular imports.
+MultipartParseOptions._DEFAULT_HANDLERS = Handlers({
+    'application/json': JSONHandler(),
+    'application/json; charset=UTF-8': JSONHandler(),
+    MEDIA_URLENCODED: URLEncodedFormHandler(),
+})
diff --git a/falcon/media/multipart.py b/falcon/media/multipart.py
new file mode 100644
index 0000000..0bacb7f
--- /dev/null
+++ b/falcon/media/multipart.py
@@ -0,0 +1,417 @@
+# Copyright 2019-2020 by Vytautas Liuolia.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Multipart form media handler."""
+
+import cgi
+import re
+from urllib.parse import unquote_to_bytes
+
+from falcon import errors
+from falcon import request_helpers
+from falcon.media.base import BaseHandler
+from falcon.util import BufferedReader, DelimiterError
+from falcon.util import misc
+
+# TODO(vytas):
+#   * Better support for form-wide charset setting
+#   * Clean up, simplify, and optimize BufferedReader
+#   * Better documentation
+
+_ALLOWED_CONTENT_HEADERS = frozenset([
+    b'content-type',
+    b'content-disposition',
+    b'content-transfer-encoding',
+])
+
+_FILENAME_STAR_RFC5987 = re.compile(r"([\w-]+)'[\w]*'(.+)")
+
+_CRLF = b'\r\n'
+_CRLF_CRLF = _CRLF + _CRLF
+
+
+class MultipartParseError(errors.HTTPBadRequest):
+    """Represents a multipart form parsing error.
+
+    This error may refer to a malformed or truncated form, usage of deprecated
+    or unsupported features, or form parameters exceeding limits configured in
+    :class:`MultipartParseOptions`.
+
+    :class:`MultipartParseError` instances raised in this module always include
+    a short human-readable description of the error.
+    """
+
+    def __init__(self, description=None, headers=None, **kwargs):
+        super().__init__('Malformed multipart/form-data request media',
+                         description, headers, **kwargs)
+
+
+# TODO(vytas): Consider supporting -charset- stuff.
+#   Does anyone use that (?)
+class BodyPart:
+    """Represents a body part in a multipart form.
+
+    Note:
+        `BodyPart` is meant to be instantiated directly only by the
+        `MultipartForm` parser.
+
+    Attributes:
+        content_type (str): Value of the Content-Type header, or the multipart
+            form default ``text/plain`` if the header is missing.
+
+        data (bytes): Body part content bytes. The maximum number of bytes that
+            could be read is configurable via :class:`MultipartParseOptions`,
+            and a :class:`.MultipartParseError` is raised if the body part is
+            larger that this size.
+
+            For large bodies, such as attached files, use the input `stream`
+            directly.
+
+            .. note::
+               Accessing this property the first time would consume the part
+               input stream.
+               The value is cached for subsequent access.
+
+        filename (str): File name if the body part is an attached file, and
+            ``None`` otherwise.
+
+        secure_filename (str): The sanitized version of `filename` using only
+            the most common ASCII characters for maximum portability and safety
+            wrt using this name as a filename on a regular file system.
+
+            If `filename` is empty or unset when referencing this property, an
+            instance of :class:`MultipartParseError` will be raised.
+
+            See also: :func:`~.secure_filename`
+
+        stream: File-like input object for reading the body part of the
+            multipart form request, if any. This object provides direct access
+            to the server's data stream and is non-seekable. The stream is
+            automatically delimited according to the multipart stream boundary.
+
+        media (object): Returns a deserialized form of the multipart body part.
+            When called, it will attempt to deserialize the body part stream
+            using the Content-Type header as well as the media-type handlers
+            configured via :class:`MultipartParseOptions`.
+
+
+        text (str): The part decoded as a text string provided the part is
+            encoded as ``text/plain``, ``None`` otherwise.
+
+            Text is decoded from `data` using the charset specified in the
+            `Content-Type` header, or, if omitted, the
+            :data:`default charset <MultipartParseOptions.default_charset>`.
+            The charset must be supported by Python's ``bytes.decode()``
+            function. The list of standard encodings (charsets) supported by
+            the Python 3 standard library can be found `here
+            <https://docs.python.org/3/library/codecs.html#standard-encodings>`__.
+
+            If decoding fails due to invalid `data` bytes (for the specified
+            encoding), or the specified encoding itself is unsupported, a
+            :class:`MultipartParseError` will be raised when referencing this
+            property.
+
+            .. note::
+               As this property builds upon `data`, it would consume the part
+               input stream in the same way.
+    """
+
+    _content_disposition = None
+    _data = None
+    _filename = None
+    _media = None
+    _name = None
+
+    def __init__(self, stream, headers, parse_options):
+        self.stream = stream
+        self._headers = headers
+        self._parse_options = parse_options
+
+    @property
+    def data(self):
+        if self._data is None:
+            max_size = self._parse_options.max_body_part_buffer_size + 1
+            self._data = self.stream.read(max_size)
+            if len(self._data) >= max_size:
+                raise MultipartParseError('body part is too large')
+
+        return self._data
+
+    @property
+    def text(self):
+        content_type, options = cgi.parse_header(self.content_type)
+        if content_type != 'text/plain':
+            return None
+
+        charset = options.get('charset', self._parse_options.default_charset)
+        try:
+            return self.data.decode(charset)
+        except (ValueError, LookupError):
+            raise MultipartParseError(
+                'invalid text or charset: {}'.format(charset))
+
+    @property
+    def content_type(self):
+        # NOTE(vytas): RFC 7578, section 4.4.
+        #   Each part MAY have an (optional) "Content-Type" header field, which
+        #   defaults to "text/plain".
+        value = self._headers.get(b'content-type', b'text/plain')
+        return value.decode('ascii')
+
+    @property
+    def filename(self):
+        if self._filename is None:
+
+            if self._content_disposition is None:
+                value = self._headers.get(b'content-disposition', b'')
+                self._content_disposition = cgi.parse_header(value.decode())
+
+            _, params = self._content_disposition
+
+            # NOTE(vytas): Supporting filename* as per RFC 5987, as that has
+            #   been spotted in the wild, even though RFC 7578 forbids it.
+            match = _FILENAME_STAR_RFC5987.match(params.get('filename*', ''))
+            if match:
+                charset, value = match.groups()
+                try:
+                    self._filename = unquote_to_bytes(value).decode(charset)
+                except (ValueError, LookupError):
+                    raise MultipartParseError(
+                        'invalid text or charset: {}'.format(charset))
+            else:
+                value = params.get('filename')
+                if value is None:
+                    return None
+                self._filename = value
+
+        return self._filename
+
+    @property
+    def secure_filename(self):
+        try:
+            return misc.secure_filename(self.filename)
+        except ValueError as ex:
+            raise MultipartParseError(str(ex))
+
+    @property
+    def name(self):
+        if self._name is None:
+
+            if self._content_disposition is None:
+                value = self._headers.get(b'content-disposition', b'')
+                self._content_disposition = cgi.parse_header(value.decode())
+
+            _, params = self._content_disposition
+            self._name = params.get('name')
+
+        return self._name
+
+    @property
+    def media(self):
+        if self._media is None:
+            handler = self._parse_options.media_handlers.find_by_media_type(
+                self.content_type, 'text/plain')
+            self._media = handler.deserialize(
+                self.stream, self.content_type, None)
+
+        return self._media
+
+
+class MultipartForm:
+
+    def __init__(self, stream, boundary, content_length, parse_options):
+        # NOTE(vytas): More lenient check whether the provided stream is not
+        #   already an instance of BufferedReader.
+        # This approach makes testing both the Cythonized and pure-Python
+        #   streams easier within the same test/benchmark suite.
+        if not hasattr(stream, 'read_until'):
+            if isinstance(stream, request_helpers.BoundedStream):
+                stream = BufferedReader(stream.stream.read, content_length)
+            else:
+                stream = BufferedReader(stream.read, content_length)
+
+        self._stream = stream
+        self._boundary = boundary
+        # NOTE(vytas): Here self._dash_boundary is not prepended with CRLF
+        #   (yet) for parsing the prologue. The CRLF will be prepended later to
+        #   construct the inter-part delimiter as per RFC 7578, section 4.1
+        #   (see the note below).
+        self._dash_boundary = b'--' + boundary
+        self._parse_options = parse_options
+
+    def __iter__(self):
+        prologue = True
+        delimiter = self._dash_boundary
+        stream = self._stream
+        max_headers_size = self._parse_options.max_body_part_headers_size
+        remaining_parts = self._parse_options.max_body_part_count
+
+        while True:
+            # NOTE(vytas): Either exhaust the unused stream part, or skip
+            #   the prologue.
+            try:
+                stream.pipe_until(delimiter, consume_delimiter=True)
+
+                if prologue:
+                    # NOTE(vytas): RFC 7578, section 4.1.
+                    #   As with other multipart types, the parts are delimited
+                    #   with a boundary delimiter, constructed using CRLF,
+                    #   "--", and the value of the "boundary" parameter.
+                    delimiter = _CRLF + delimiter
+                    prologue = False
+
+                separator = stream.read_until(_CRLF, 2, consume_delimiter=True)
+                if separator == b'--':
+                    # NOTE(vytas): boundary delimiter + '--\r\n' signals the
+                    # end of a multipart form.
+                    break
+                elif separator:
+                    raise MultipartParseError('unexpected form structure')
+
+            except DelimiterError:
+                raise MultipartParseError('unexpected form structure')
+
+            headers = {}
+            try:
+                headers_block = stream.read_until(
+                    _CRLF_CRLF, max_headers_size, consume_delimiter=True)
+            except DelimiterError:
+                raise MultipartParseError('incomplete body part headers')
+
+            for line in headers_block.split(_CRLF):
+                name, sep, value = line.partition(b': ')
+                if sep:
+                    name = name.lower()
+
+                    # NOTE(vytas): RFC 7578, section 4.5.
+                    #   This use is deprecated for use in contexts that support
+                    #   binary data such as HTTP. Senders SHOULD NOT generate
+                    #   any parts with a Content-Transfer-Encoding header
+                    #   field.
+                    #
+                    #   Currently, no deployed implementations that send such
+                    #   bodies have been discovered.
+                    if name == b'content-transfer-encoding':
+                        raise MultipartParseError(
+                            'the deprecated Content-Transfer-Encoding header '
+                            'field is unsupported')
+                    # NOTE(vytas): RFC 7578, section 4.8.
+                    #   Other header fields MUST NOT be included and MUST be
+                    #   ignored.
+                    elif name in _ALLOWED_CONTENT_HEADERS:
+                        headers[name] = value
+
+            remaining_parts -= 1
+            if remaining_parts < 0 < self._parse_options.max_body_part_count:
+                raise MultipartParseError(
+                    'maximum number of form body parts exceeded')
+
+            yield BodyPart(stream.delimit(delimiter), headers,
+                           self._parse_options)
+
+
+class MultipartFormHandler(BaseHandler):
+    """Multipart form (content type ``multipart/form-data``) media handler.
+
+    The ``multipart/form-data`` media type for HTML5 forms is defined in
+    `RFC 7578 <https://tools.ietf.org/html/rfc7578>`_.
+
+    The multipart media type itself is defined in
+    `RFC 2046 section 5.1 <https://tools.ietf.org/html/rfc2046#section-5.1>`_.
+
+    .. note::
+       Unlike many form parsing implementations in other frameworks, this
+       handler does not consume the stream immediately. Rather, the stream is
+       consumed on-demand and parsed into individual body parts while iterating
+       over the media object.
+    """
+
+    def __init__(self, parse_options=None):
+        self.parse_options = parse_options or MultipartParseOptions()
+
+    def deserialize(self, stream, content_type, content_length):
+        _, options = cgi.parse_header(content_type)
+        try:
+            boundary = options['boundary']
+        except KeyError:
+            raise errors.HTTPInvalidHeader(
+                'No boundary specifier found in {!r}'.format(content_type),
+                'Content-Type')
+
+        # NOTE(vytas): RFC 2046, section 5.1.
+        #   If a boundary delimiter line appears to end with white space, the
+        #   white space must be presumed to have been added by a gateway, and
+        #   must be deleted.
+        boundary = boundary.rstrip()
+
+        # NOTE(vytas): RFC 2046, section 5.1.
+        #   The boundary parameter consists of 1 to 70 characters from a set of
+        #   characters known to be very robust through mail gateways, and NOT
+        #   ending with white space.
+        if not 1 <= len(boundary) <= 70:
+            raise errors.HTTPInvalidHeader(
+                'The boundary parameter must consist of 1 to 70 characters',
+                'Content-Type')
+
+        return MultipartForm(stream, boundary.encode(), content_length,
+                             self.parse_options)
+
+    def serialize(self, media, content_type):
+        raise NotImplementedError('multipart form serialization unsupported')
+
+
+# PERF(vytas): To avoid typos and improve storage space and speed over a dict.
+#   Inspired by RequestOptions.
+class MultipartParseOptions:
+    """Defines a set of configurable multipart form parser options.
+
+    Attributes:
+        default_charset (str): The default character encoding for
+            :data:`text fields <BodyPart.text>` (default: ``utf-8``).
+
+        max_body_part_count (int): The maximum number of body parts in the form
+            (default: 64). If the form contains more parts than this number,
+            an instance of :class:`MultipartParseError` will be raised. If this
+            option is set to 0, no limit will be imposed by the parser.
+
+        max_body_part_buffer_size (int): The maximum number of bytes to buffer
+            and return when the :data:`BodyPart.data` property is
+            referenced (default: 1 MiB). If the body part size exceeds this
+            value, an instance of :class:`MultipartParseError` will be raised.
+
+        max_body_part_headers_size (int): The maximum size (in bytes) of the
+            body part headers structure (default: 8192).
+
+        media_handlers (Handlers): A dict-like object for configuring the
+            media-types to handle. By default, handlers are provided for the
+            ``application/json`` and ``application/x-www-form-urlencoded``
+            media types.
+    """
+
+    _DEFAULT_HANDLERS = None
+
+    __slots__ = (
+        'default_charset',
+        'max_body_part_buffer_size',
+        'max_body_part_count',
+        'max_body_part_headers_size',
+        'media_handlers',
+    )
+
+    def __init__(self):
+        self.default_charset = 'utf-8'
+        self.max_body_part_buffer_size = 1024 * 1024
+        self.max_body_part_count = 64
+        self.max_body_part_headers_size = 8192
+        self.media_handlers = self._DEFAULT_HANDLERS
diff --git a/falcon/request.py b/falcon/request.py
index 8e40ded..4349155 100644
--- a/falcon/request.py
+++ b/falcon/request.py
@@ -983,7 +983,8 @@ class Request:
                 self.content_length
             )
         finally:
-            self.bounded_stream.exhaust()
+            if handler.exhaust_stream:
+                self.bounded_stream.exhaust()
 
         return self._media
 
@@ -1874,10 +1875,10 @@ class RequestOptions:
             independently, this will default to
             :attr:`falcon.DEFAULT_MEDIA_TYPE`.
 
-        media_handlers (Handlers): A dict-like object that allows you to
-            configure the media-types that you would like to handle.
-            By default, a handler is provided for the ``application/json``
-            media type.
+        media_handlers (Handlers): A dict-like object for configuring the
+            media-types to handle. By default, handlers are provided for the
+            ``application/json``, ``application/x-www-form-urlencoded`` and
+            ``multipart/form-data`` media types.
     """
     __slots__ = (
         'keep_blank_qs_values',
diff --git a/falcon/response.py b/falcon/response.py
index 12a9ba9..3618319 100644
--- a/falcon/response.py
+++ b/falcon/response.py
@@ -1048,10 +1048,10 @@ class ResponseOptions:
             however, if created independently, this will default to
             :attr:`falcon.DEFAULT_MEDIA_TYPE`..
 
-        media_handlers (Handlers): A dict-like object that allows you to
-            configure the media-types that you would like to handle.
-            By default, a handler is provided for the ``application/json``
-            media type.
+        media_handlers (Handlers): A dict-like object for configuring the
+            media-types to handle. By default, handlers are provided for the
+            ``application/json``, ``application/x-www-form-urlencoded`` and
+            ``multipart/form-data`` media types.
 
         static_media_types (dict): A mapping of dot-prefixed file extensions to
             Internet media types (RFC 2046). Defaults to ``mimetypes.types_map``
diff --git a/falcon/util/__init__.py b/falcon/util/__init__.py
index c4362b4..41250c5 100644
--- a/falcon/util/__init__.py
+++ b/falcon/util/__init__.py
@@ -22,10 +22,13 @@ Conversely, the `uri` module must be imported explicitly::
 
 from http import cookies as http_cookies
 import json  # NOQA
+import sys
 
 # Hoist misc. utils
-from falcon.util.structures import *  # NOQA
 from falcon.util.misc import *  # NOQA
+from falcon.util.reader import BufferedReader as _PyBufferedReader
+from falcon.util.reader import DelimiterError  # NOQA
+from falcon.util.structures import *  # NOQA
 from falcon.util.sync import *  # NOQA
 from falcon.util.time import *  # NOQA
 
@@ -37,3 +40,15 @@ from falcon.util.time import *  # NOQA
 _reserved_cookie_attrs = http_cookies.Morsel._reserved  # type: ignore
 if 'samesite' not in _reserved_cookie_attrs:  # pragma: no cover
     _reserved_cookie_attrs['samesite'] = 'SameSite'  # type: ignore
+
+
+IS_64_BITS = sys.maxsize > 2**32
+
+try:
+    from falcon.cyutil.reader import BufferedReader as _CyBufferedReader
+except ImportError:
+    _CyBufferedReader = None
+
+# NOTE(vytas): Cythonized BufferedReader makes heavy use of Py_ssize_t which
+#   would overflow on 32-bit systems with form parts larger than 2 GiB.
+BufferedReader = (_CyBufferedReader or _PyBufferedReader) if IS_64_BITS else _PyBufferedReader
diff --git a/falcon/util/misc.py b/falcon/util/misc.py
index d9962a8..5cf4699 100644
--- a/falcon/util/misc.py
+++ b/falcon/util/misc.py
@@ -29,7 +29,9 @@ import functools
 import http
 import inspect
 import os
+import re
 import sys
+import unicodedata
 import warnings
 
 from falcon import status_codes
@@ -46,8 +48,10 @@ __all__ = (
     'get_http_status',
     'http_status_to_code',
     'code_to_http_status',
+    'secure_filename',
 )
 
+_UNSAFE_CHARS = re.compile(r'[^a-zA-Z0-9.-]')
 
 # PERF(kgriffs): Avoid superfluous namespace lookups
 strptime = datetime.datetime.strptime
@@ -363,6 +367,46 @@ def get_http_status(status_code, default_reason='Unknown'):
         return str(code) + ' ' + default_reason
 
 
+def secure_filename(filename):
+    """Sanitize the provided `filename` to contain only ASCII characters.
+
+    Only ASCII alphanumerals, ``'.'``, ``'-'`` and ``'_'`` are allowed for
+    maximum portability and safety wrt using this name as a filename on a
+    regular file system. All other characters will be replaced with an
+    underscore (``'_'``).
+
+    .. note::
+        The `filename` is normalized to the Unicode ``NKFD`` form prior to
+        ASCII conversion in order to extract more alphanumerals where a
+        decomposition is available. For instance:
+
+        >>> secure_filename('Bold Digit 𝟏')
+        'Bold_Digit_1'
+        >>> secure_filename('Ångström unit physics.pdf')
+        'A_ngstro_m_unit_physics.pdf'
+
+    Args:
+        filename (str): Arbitrary filename input from the request, such as a
+            multipart form filename field.
+
+    Returns:
+        str: The sanitized filename.
+
+    Raises:
+        ValueError: the provided filename is an empty string.
+    """
+    # TODO(vytas): max_length (int): Maximum length of the returned
+    #     filename. Should the returned filename exceed this restriction, it is
+    #     truncated while attempting to preserve the extension.
+    if not filename:
+        raise ValueError('filename may not be an empty string')
+
+    filename = unicodedata.normalize('NFKD', filename)
+    if filename.startswith('.'):
+        filename = filename.replace('.', '_', 1)
+    return _UNSAFE_CHARS.sub('_', filename)
+
+
 @_lru_cache_safe(maxsize=64)
 def http_status_to_code(status):
     """Normalize an HTTP status to an integer code.
diff --git a/falcon/util/reader.py b/falcon/util/reader.py
new file mode 100644
index 0000000..f51e3f3
--- /dev/null
+++ b/falcon/util/reader.py
@@ -0,0 +1,384 @@
+# Copyright 2019-2020 by Vytautas Liuolia.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Buffered stream reader."""
+
+import functools
+import io
+
+DEFAULT_CHUNK_SIZE = 32768
+"""Default chunk size for :class:`BufferedReader` (32 KiB)."""
+
+_MAX_JOIN_CHUNKS = 128
+
+
+class DelimiterError(IOError):
+    pass
+
+
+class BufferedReader:
+
+    def __init__(self, read, max_stream_len, chunk_size=None):
+        self._read_func = read
+        self._chunk_size = chunk_size or DEFAULT_CHUNK_SIZE
+        self._max_join_size = self._chunk_size * _MAX_JOIN_CHUNKS
+
+        self._buffer = b''
+        self._buffer_len = 0
+        self._buffer_pos = 0
+        self._max_bytes_remaining = max_stream_len
+
+    def _perform_read(self, size):
+        # PERF(vytas): In Cython, bind types:
+        #   cdef bytes chunk
+        #   cdef Py_ssize_t chunk_len
+        #   cdef result
+
+        size = min(size, self._max_bytes_remaining)
+        if size <= 0:
+            return b''
+
+        chunk = self._read_func(size)
+        chunk_len = len(chunk)
+        self._max_bytes_remaining -= chunk_len
+        if chunk_len == size:
+            return chunk
+
+        if chunk_len == 0:
+            # NOTE(vytas): The EOF.
+            self._max_bytes_remaining = 0
+            return b''
+
+        result = io.BytesIO(chunk)
+        result.seek(chunk_len)
+
+        while True:
+            size -= chunk_len
+            if size <= 0:
+                return result.getvalue()
+
+            chunk = self._read_func(size)
+            chunk_len = len(chunk)
+            if chunk_len == 0:
+                # NOTE(vytas): The EOF.
+                self._max_bytes_remaining = 0
+                return result.getvalue()
+
+            self._max_bytes_remaining -= chunk_len
+            result.write(chunk)
+
+    def _fill_buffer(self):
+        # PERF(vytas): In Cython, bind types:
+        #   cdef Py_ssize_t read_size
+
+        if self._buffer_len - self._buffer_pos < self._chunk_size:
+            read_size = self._chunk_size - (self._buffer_len - self._buffer_pos)
+
+            if self._buffer_pos == 0:
+                self._buffer += self._perform_read(read_size)
+            else:
+                self._buffer = (self._buffer[self._buffer_pos:] +
+                                self._perform_read(read_size))
+                self._buffer_pos = 0
+
+            self._buffer_len = len(self._buffer)
+
+    def peek(self, size=-1):
+        if size < 0 or size > self._chunk_size:
+            size = self._chunk_size
+
+        if self._buffer_len - self._buffer_pos < size:
+            self._fill_buffer()
+
+        return self._buffer[self._buffer_pos:self._buffer_pos + size]
+
+    def _normalize_size(self, size):
+        # PERF(vytas): In Cython, bind types:
+        #   cdef Py_ssize_t result
+        #   cdef Py_ssize_t max_size
+
+        max_size = (self._max_bytes_remaining + self._buffer_len -
+                    self._buffer_pos)
+
+        if size is None or size == -1 or size > max_size:
+            return max_size
+        return size
+
+    def read(self, size=-1):
+        return self._read(self._normalize_size(size))
+
+    def _read(self, size):
+        # PERF(vytas): In Cython, bind types:
+        #   cdef Py_ssize_t read_size
+        #   cdef bytes result
+
+        # NOTE(vytas): Dish directly from the buffer, if possible.
+        if size <= self._buffer_len - self._buffer_pos:
+            if size == self._buffer_len and self._buffer_pos == 0:
+                result = self._buffer
+                self._buffer_len = 0
+                self._buffer = b''
+                return result
+
+            self._buffer_pos += size
+            return self._buffer[self._buffer_pos - size:self._buffer_pos]
+
+        # NOTE(vytas): Pass through large reads.
+        if self._buffer_len == 0 and size >= self._chunk_size:
+            return self._perform_read(size)
+
+        # NOTE(vytas): if size > self._buffer_len - self._buffer_pos
+        read_size = size - (self._buffer_len - self._buffer_pos)
+        result = self._buffer[self._buffer_pos:]
+
+        if read_size >= self._chunk_size:
+            self._buffer_len = 0
+            self._buffer_pos = 0
+            self._buffer = b''
+            return result + self._perform_read(read_size)
+
+        self._buffer = self._perform_read(self._chunk_size)
+        self._buffer_len = len(self._buffer)
+        self._buffer_pos = read_size
+        return result + self._buffer[:read_size]
+
+    def read_until(self, delimiter, size=-1, consume_delimiter=False):
+        # PERF(vytas): In Cython, bind types:
+        #   cdef Py_ssize_t read_size
+        #   cdef result
+
+        read_size = self._normalize_size(size)
+        if read_size <= self._max_join_size:
+            return self._read_until(delimiter, read_size, consume_delimiter)
+
+        # NOTE(vytas): A large size was requested, optimize for memory
+        #   consumption by avoiding to momentarily keep both the chunks and the
+        #   joint result in memory at the same time.
+        result = io.BytesIO()
+        self.pipe_until(delimiter, result, consume_delimiter, read_size)
+        return result.getvalue()
+
+    def _finalize_read_until(
+            self, size, backlog, have_bytes, consume_bytes, delimiter=None,
+            delimiter_pos=-1, next_chunk=None, next_chunk_len=0):
+
+        if delimiter_pos < 0 and delimiter is not None:
+            delimiter_pos = self._buffer.find(delimiter, self._buffer_pos)
+
+        if delimiter_pos >= 0:
+            size = min(size, have_bytes + delimiter_pos - self._buffer_pos)
+
+        if have_bytes == 0:
+            # PERF(vytas): Do not join bytes unless needed.
+            ret_value = self._read(size)
+        else:
+            backlog.append(self._read(size - have_bytes))
+            ret_value = b''.join(backlog)
+
+        if next_chunk_len > 0:
+            if self._buffer_len == 0:
+                self._buffer = next_chunk
+                self._buffer_len = next_chunk_len
+            else:
+                self._buffer = self._buffer[self._buffer_pos:] + next_chunk
+                self._buffer_len = (self._buffer_len - self._buffer_pos +
+                                    next_chunk_len)
+                self._buffer_pos = 0
+
+        if consume_bytes:
+            if delimiter_pos < 0:
+                if self.peek(consume_bytes) != delimiter:
+                    raise DelimiterError('expected delimiter missing')
+            elif self._buffer_pos != delimiter_pos:
+                # NOTE(vytas): If we are going to consume the delimiter the
+                #   quick way (i.e., skipping the above peek() check), we must
+                #   make sure it is directly succeeding the result.
+                raise DelimiterError('expected delimiter missing')
+
+            self._buffer_pos += consume_bytes
+
+        return ret_value
+
+    def _read_until(self, delimiter, size, consume_delimiter):
+        # PERF(vytas): In Cython, bind types:
+        #   cdef list result = []
+        #   cdef Py_ssize_t have_bytes = 0
+        #   cdef Py_ssize_t delimiter_len_1 = len(delimiter) - 1
+        #   cdef Py_ssize_t delimiter_pos = -1
+        #   cdef Py_ssize_t consume_bytes
+        #   cdef Py_ssize_t offset
+
+        result = []
+        have_bytes = 0
+        delimiter_len_1 = len(delimiter) - 1
+        delimiter_pos = -1
+        consume_bytes = (delimiter_len_1 + 1) if consume_delimiter else 0
+
+        if not 0 <= delimiter_len_1 < self._chunk_size:
+            raise ValueError('delimiter length must be within [1, chunk_size]')
+
+        # PERF(vytas): If the requested size is equal to the chunk size (or is
+        #   a multiple of it), align the buffer.
+        #   This can often nearly *double* the performance if one is reading in
+        #   chunks.
+        if size % self._chunk_size == 0:
+            self._fill_buffer()
+
+        while True:
+            if self._buffer_len > self._buffer_pos:
+                delimiter_pos = self._buffer.find(delimiter, self._buffer_pos)
+                if delimiter_pos >= 0:
+                    # NOTE(vytas): Delimiter was found in the current buffer.
+                    #   We can now return to the caller.
+                    return self._finalize_read_until(
+                        size, result, have_bytes, consume_bytes,
+                        delimiter_pos=delimiter_pos)
+
+            if size < (have_bytes + self._buffer_len - self._buffer_pos -
+                       delimiter_len_1):
+                # NOTE(vytas): We now have enough data in the buffer to return
+                #   to the caller.
+                return self._finalize_read_until(
+                    size, result, have_bytes, consume_bytes, delimiter)
+
+            next_chunk = self._perform_read(self._chunk_size)
+            next_chunk_len = len(next_chunk)
+            if self._max_bytes_remaining == 0:
+                # NOTE(vytas): We have reached the EOF.
+                self._buffer_len += next_chunk_len
+                self._buffer += next_chunk
+                return self._finalize_read_until(
+                    size, result, have_bytes, consume_bytes, delimiter)
+
+            # NOTE(vytas): The buffer was empty before, skip straight to the
+            #   next chunk.
+            if self._buffer_len <= self._buffer_pos:
+                self._buffer_len = next_chunk_len
+                self._buffer_pos = 0
+                self._buffer = next_chunk
+                continue
+
+            # NOTE(vytas): We must check there is no delimiter in the chunk
+            #   boundary before we can safely splice them.
+            if delimiter_len_1 > 0:
+                offset = max(self._buffer_len - delimiter_len_1,
+                             self._buffer_pos)
+                fragment = (self._buffer[offset:] +
+                            next_chunk[:delimiter_len_1])
+                delimiter_pos = fragment.find(delimiter)
+                if delimiter_pos >= 0:
+                    self._buffer_len += next_chunk_len
+                    self._buffer += next_chunk
+                    return self._finalize_read_until(
+                        size, result, have_bytes, consume_bytes, delimiter,
+                        delimiter_pos + offset)
+
+            if have_bytes + self._buffer_len - self._buffer_pos >= size:
+                # NOTE(vytas): we have now verified that all bytes currently in
+                #   the buffer are delimiter-free, including the border of the
+                #   upcoming chunk
+                return self._finalize_read_until(
+                    size, result, have_bytes, consume_bytes, delimiter,
+                    next_chunk=next_chunk, next_chunk_len=next_chunk_len)
+
+            have_bytes += self._buffer_len - self._buffer_pos
+            if self._buffer_pos > 0:
+                result.append(self._buffer[self._buffer_pos:])
+            else:
+                result.append(self._buffer)
+            self._buffer_len = next_chunk_len
+            self._buffer_pos = 0
+            self._buffer = next_chunk
+
+    def pipe(self, destination=None):
+        while True:
+            chunk = self.read(self._chunk_size)
+            if not chunk:
+                break
+
+            if destination is not None:
+                destination.write(chunk)
+
+    def pipe_until(self, delimiter, destination=None, consume_delimiter=False,
+                   _size=None):
+        # PERF(vytas): In Cython, bind types:
+        #   cdef Py_ssize_t remaining
+
+        remaining = self._normalize_size(_size)
+
+        while remaining > 0:
+            chunk = self._read_until(
+                delimiter, min(self._chunk_size, remaining), False)
+            if not chunk:
+                break
+
+            if destination is not None:
+                destination.write(chunk)
+
+            remaining -= self._chunk_size
+
+        if consume_delimiter:
+            delimiter_len = len(delimiter)
+            if self.peek(delimiter_len) != delimiter:
+                raise DelimiterError('expected delimiter missing')
+            self._buffer_pos += delimiter_len
+
+    def exhaust(self):
+        self.pipe()
+
+    def delimit(self, delimiter):
+        read = functools.partial(self.read_until, delimiter)
+        return type(self)(read, self._normalize_size(None), self._chunk_size)
+
+    def readline(self, size=-1):
+        size = self._normalize_size(size)
+
+        result = self.read_until(b'\n', size)
+        if len(result) < size:
+            return result + self.read(1)
+        return result
+
+    def readlines(self, hint=-1):
+        # PERF(vytas): In Cython, bind types:
+        #   cdef Py_ssize_t read
+        #   cdef list result = []
+        read = 0
+        result = []
+
+        while True:
+            line = self.readline()
+            if not line:
+                break
+            result.append(line)
+
+            if hint >= 0:
+                read += len(line)
+                if read >= hint:
+                    break
+
+        return result
+
+    # --- implementing IOBase methods, the duck-typing way ---
+
+    def readable(self):
+        """Always returns ``True``."""
+        return True
+
+    def seekable(self):
+        """Always returns ``False``."""
+        return False
+
+    def writeable(self):
+        """Always returns ``False``."""
+        return False
diff --git a/setup.py b/setup.py
index 9a88ccb..641cc86 100644
--- a/setup.py
+++ b/setup.py
@@ -84,6 +84,11 @@ if CYTHON:
         if (package + '.' + module) not in modules_to_exclude
     ]
 
+    # NOTE(vytas): Now that all our codebase is Python 3.5+, specify the
+    #   Python 3 language level for Cython as well to avoid any surprises.
+    for ext_mod in ext_modules:
+        ext_mod.cython_directives = {'language_level': '3'}
+
     cmdclass = {'build_ext': build_ext}
 
 else:


Test patch:
diff --git a/tests/test_buffered_reader.py b/tests/test_buffered_reader.py
new file mode 100644
index 0000000..f972798
--- /dev/null
+++ b/tests/test_buffered_reader.py
@@ -0,0 +1,350 @@
+import io
+
+import pytest
+
+from falcon.util import BufferedReader, DelimiterError
+
+
+class WouldHang(RuntimeError):
+    pass
+
+
+class GlitchyStream(io.BytesIO):
+
+    def read(self, size=None):
+        if size is None or size == -1:
+            raise WouldHang('unbounded read()')
+
+        result = super().read(size)
+        if not result:
+            raise WouldHang('EOF')
+        return result
+
+
+class FragmentedStream(GlitchyStream):
+    def read(self, size=None):
+        if size is None or size <= 1:
+            return super().read(size)
+
+        if size < 8:
+            size = 1
+        else:
+            size = size // 2
+
+        return super().read(size)
+
+
+TEST_DATA = (
+    b'123456789ABCDEF\n' * 64 * 1024 * 64 +
+    b'--boundary1234567890--' +
+    b'123456789ABCDEF\n' * 64 * 1024 * 63 +
+    b'--boundary1234567890--' +
+    b'123456789ABCDEF\n' * 64 * 1024 * 62 +
+    b'--boundary1234567890--'
+)
+
+TEST_BYTES_IO = GlitchyStream(TEST_DATA)
+
+
+@pytest.fixture()
+def buffered_reader():
+    def _reader_fixture(chunk_size=None):
+        TEST_BYTES_IO.seek(0)
+        return BufferedReader(TEST_BYTES_IO.read, len(TEST_DATA), chunk_size)
+
+    yield _reader_fixture
+
+
+@pytest.fixture()
+def shorter_stream():
+    TEST_BYTES_IO.seek(0)
+    return BufferedReader(TEST_BYTES_IO.read, 1024, 128)
+
+
+@pytest.fixture()
+def fragmented_stream():
+    stream = FragmentedStream(TEST_DATA)
+    return BufferedReader(stream.read, len(TEST_DATA))
+
+
+def test_peek(buffered_reader):
+    stream = buffered_reader(16)
+
+    assert stream.peek(0) == b''
+    assert stream.peek(1) == b'1'
+    assert stream.peek(2) == b'12'
+    assert stream.peek(16) == b'123456789ABCDEF\n'
+    assert stream.peek(17) == b'123456789ABCDEF\n'
+    assert stream.peek(32) == b'123456789ABCDEF\n'
+
+    assert stream.read(15) == b'123456789ABCDEF'
+
+    assert stream.peek(0) == b''
+    assert stream.peek(1) == b'\n'
+    assert stream.peek(2) == b'\n1'
+    assert stream.peek(16) == b'\n123456789ABCDEF'
+    assert stream.peek(17) == b'\n123456789ABCDEF'
+    assert stream.peek(32) == b'\n123456789ABCDEF'
+
+
+def test_peek_eof():
+    source = b'Hello, world!\n'
+    source_stream = io.BytesIO(source)
+    stream = BufferedReader(source_stream.read, len(source) - 1)
+
+    assert stream.peek(0) == b''
+    assert stream.peek(1) == b'H'
+    assert stream.peek(2) == b'He'
+    assert stream.peek(16) == b'Hello, world!'
+    assert stream.peek(32) == b'Hello, world!'
+
+    assert source_stream.read() == b'\n'
+
+
+def test_bounded_read():
+    stream = io.BytesIO(b'Hello, world!')
+    buffered = BufferedReader(stream.read, len('Hello, world'))
+    buffered.read()
+
+    assert stream.read() == b'!'
+
+
+@pytest.mark.parametrize('size', [
+    0,
+    1,
+    2,
+    7,
+    62,
+    63,
+    64,
+    65,
+    126,
+    127,
+    128,
+    129,
+    1000,
+    10000,
+])
+def test_read_from_buffer(buffered_reader, size):
+    stream = buffered_reader(64)
+    stream.peek(64)
+
+    assert stream.read(size) == TEST_DATA[:size]
+    assert stream.read(1) == TEST_DATA[size:size + 1]
+
+
+def test_read_until_delimiter_size_check(buffered_reader):
+    stream = buffered_reader(64)
+
+    with pytest.raises(ValueError):
+        stream.read_until(b'')
+    with pytest.raises(ValueError):
+        stream.read_until(b'B' * 65)
+
+
+@pytest.mark.parametrize('size', [
+    0,
+    1,
+    2,
+    7,
+    62,
+    63,
+    64,
+    65,
+    126,
+    127,
+    128,
+    129,
+    1000,
+    10000,
+])
+def test_read_until_with_size(buffered_reader, size):
+    stream = buffered_reader(64)
+    assert stream.read_until(b'--boundary1234567890--', size) == (
+        TEST_DATA[:size])
+
+
+def test_read_until(buffered_reader):
+    stream = buffered_reader()
+
+    assert len(stream.read_until(b'--boundary1234567890--')) == 64 * 1024**2
+    stream.read_until(b'123456789ABCDEF\n')
+    assert len(stream.read_until(b'--boundary1234567890--')) == 63 * 1024**2
+    stream.read_until(b'123456789ABCDEF\n')
+    assert len(stream.read_until(b'--boundary1234567890--')) == 62 * 1024**2
+
+
+@pytest.mark.parametrize('size1,size2', [
+    (11003077, 22000721),
+    (13372477, 51637898),
+])
+def test_irregular_large_read_until(buffered_reader, size1, size2):
+    stream = buffered_reader()
+    delimiter = b'--boundary1234567890--'
+
+    stream.pipe_until(delimiter, consume_delimiter=True)
+    stream.pipe_until(delimiter, consume_delimiter=True)
+
+    expected = b'123456789ABCDEF\n' * 64 * 1024 * 62
+
+    assert stream.read_until(delimiter, 1337) == expected[:1337]
+
+    chunk1 = stream.read_until(delimiter, size1)
+    assert len(chunk1) == size1
+
+    chunk2 = stream.read_until(delimiter, size2)
+    assert len(chunk2) == size2
+
+    remainder = stream.read_until(delimiter, 62 * 1024 * 1024)
+    assert chunk1 + chunk2 + remainder == expected[1337:]
+
+
+@pytest.mark.parametrize('size', [
+    0,
+    1,
+    2,
+    7,
+    62,
+    63,
+    64,
+    65,
+    126,
+    127,
+    128,
+    129,
+    1000,
+])
+def test_read_until_from_buffer(shorter_stream, size):
+    shorter_stream.peek(128)
+
+    assert shorter_stream.read_until(b'\n1', size) == (
+        b'123456789ABCDEF'[:size])
+
+
+def test_read_until_missing_delimiter(shorter_stream):
+    class BoundaryMissing(RuntimeError):
+        pass
+
+    with pytest.raises(DelimiterError):
+        shorter_stream.read_until(b'--boundary1234567890--',
+                                  consume_delimiter=True)
+
+
+def test_consume_delimiter(shorter_stream):
+    shorter_stream.peek()
+    shorter_stream.read(113)
+
+    assert shorter_stream.peek(1) == b'2'
+
+    chunk = shorter_stream.read_until(b'1', 15, consume_delimiter=True)
+    assert chunk == b'23456789ABCDEF\n'
+
+    assert shorter_stream.read_until(b'ABCDE', 4) == b'2345'
+
+
+@pytest.mark.parametrize('chunk_size', list(range(46, 63)))
+def test_read_until_shared_boundary(chunk_size):
+    source = b'-boundary-like-' * 4 + b'--some junk--\n' + b'\n' * 1024
+    source_stream = io.BytesIO(source)
+    stream = BufferedReader(source_stream.read, len(source), chunk_size)
+    assert stream.read_until(b'-boundary-like---') == b'-boundary-like-' * 3
+    assert stream.peek(17) == b'-boundary-like---'
+
+
+def test_pipe(shorter_stream):
+    output = io.BytesIO()
+    shorter_stream.pipe(output)
+    assert len(output.getvalue()) == 1024
+
+
+def test_pipe_until(buffered_reader):
+    stream = buffered_reader(2 ** 16)
+
+    output = io.BytesIO()
+    stream.pipe_until(b'--boundary1234567890--', output)
+    assert len(output.getvalue()) == 64 * 1024**2
+
+
+def test_pipe_until_without_destination(buffered_reader):
+    stream = buffered_reader(2 ** 16)
+    stream.pipe_until(b'--boundary1234567890--')
+    assert stream.peek(22) == b'--boundary1234567890--'
+
+
+def test_exhaust(shorter_stream):
+    shorter_stream.exhaust()
+    assert not shorter_stream.read()
+
+
+def test_readline():
+    source = (
+        b'Hello, world!\n'
+        b'A line.\n'
+        b'\n'
+        b'A longer line... \n' +
+        b'SPAM ' * 7 + b'\n' +
+        b'\n'
+    )
+
+    stream = BufferedReader(io.BytesIO(source).read, len(source))
+    assert stream.readline() == b'Hello, world!\n'
+    assert stream.readline() == b'A line.\n'
+    assert stream.readline() == b'\n'
+    assert stream.readline() == b'A longer line... \n'
+    assert stream.readline() == b'SPAM SPAM SPAM SPAM SPAM SPAM SPAM \n'
+    assert stream.readline() == b'\n'
+    assert stream.readline() == b''
+    assert stream.readline() == b''
+
+
+def test_readline_with_size():
+    source = (
+        b'Hello, world! This is a short village name in Wales.\n'
+        b'Llanfairpwllgwyngyllgogerychwyrndrobwllllantysiliogogogoch')
+
+    stream = BufferedReader(io.BytesIO(source).read, len(source))
+    assert stream.readline(37) == b'Hello, world! This is a short village'
+    assert stream.readline(37) == b' name in Wales.\n'
+    assert stream.readline(8) == b'Llanfair'
+    assert stream.readline(16) == b'pwllgwyngyllgoge'
+    assert stream.readline(64) == b'rychwyrndrobwllllantysiliogogogoch'
+
+
+def test_readlines(shorter_stream):
+    assert shorter_stream.readlines() == [b'123456789ABCDEF\n'] * 64
+
+
+@pytest.mark.parametrize('chunk_size', [
+    8,
+    16,
+    256,
+    1024,
+    2 ** 16,
+])
+def test_readlines_hint(buffered_reader, chunk_size):
+    stream = buffered_reader(chunk_size)
+    assert stream.readlines(100) == [b'123456789ABCDEF\n'] * 7
+    assert stream.readlines(64) == [b'123456789ABCDEF\n'] * 4
+    assert stream.readlines(16) == [b'123456789ABCDEF\n']
+
+
+def test_duck_compatibility_with_io_base(shorter_stream):
+    assert shorter_stream.readable()
+    assert not shorter_stream.seekable()
+    assert not shorter_stream.writeable()
+
+
+def test_fragmented_reads(fragmented_stream):
+    b = io.BytesIO()
+    fragmented_stream.pipe_until(b'--boundary1234567890--', b)
+    assert fragmented_stream.read(2) == b'--'
+
+    fragmented_stream.pipe_until(b'--boundary1234567890--')
+    assert fragmented_stream.read(3) == b'--b'
+
+    fragmented_stream.pipe_until(b'--boundary1234567890--')
+    assert fragmented_stream.read(7) == b'--bound'
+
+    fragmented_stream.exhaust()
+    assert fragmented_stream.read(4) == b''
+    assert fragmented_stream.read() == b''
diff --git a/tests/test_cython.py b/tests/test_cython.py
index 92c8210..ad687f9 100644
--- a/tests/test_cython.py
+++ b/tests/test_cython.py
@@ -1,6 +1,9 @@
+import io
+
 import pytest
 
 import falcon
+import falcon.util
 
 try:
     import cython
@@ -13,3 +16,11 @@ class TestCythonized:
     @pytest.mark.skipif(not cython, reason='Cython not installed')
     def test_imported_from_c_modules(self):
         assert 'falcon/app.py' not in str(falcon.app)
+
+    def test_stream_has_private_read(self):
+        stream = falcon.util.BufferedReader(io.BytesIO().read, 8)
+
+        if cython and falcon.util.IS_64_BITS:
+            assert not hasattr(stream, '_read')
+        else:
+            assert hasattr(stream, '_read')
diff --git a/tests/test_media_multipart.py b/tests/test_media_multipart.py
new file mode 100644
index 0000000..8683bfb
--- /dev/null
+++ b/tests/test_media_multipart.py
@@ -0,0 +1,608 @@
+import io
+import itertools
+import os
+import random
+
+import pytest
+
+import falcon
+from falcon import media
+from falcon import testing
+from falcon.util import BufferedReader
+
+
+EXAMPLE1 = (
+    b'--5b11af82ab65407ba8cdccf37d2a9c4f\r\n'
+    b'Content-Disposition: form-data; name="hello"\r\n\r\n'
+    b'world\r\n'
+    b'--5b11af82ab65407ba8cdccf37d2a9c4f\r\n'
+    b'Content-Disposition: form-data; name="document"\r\n'
+    b'Content-Type: application/json\r\n\r\n'
+    b'{"debug": true, "message": "Hello, world!", "score": 7}\r\n'
+    b'--5b11af82ab65407ba8cdccf37d2a9c4f\r\n'
+    b'Content-Disposition: form-data; name="file1"; filename="test.txt"\r\n'
+    b'Content-Type: text/plain\r\n\r\n'
+    b'Hello, world!\n\r\n'
+    b'--5b11af82ab65407ba8cdccf37d2a9c4f--\r\n')
+
+EXAMPLE2 = (
+    b'-----------------------------1574247108204320607285918568\r\n'
+    b'Content-Disposition: form-data; name="description"\r\n\r\n'
+    b'\r\n'
+    b'-----------------------------1574247108204320607285918568\r\n'
+    b'Content-Disposition: form-data; name="moderation"\r\n\r\n'
+    b'approved\r\n'
+    b'-----------------------------1574247108204320607285918568\r\n'
+    b'Content-Disposition: form-data; name="title"\r\n\r\n'
+    b'A simple text file example.\r\n'
+    b'-----------------------------1574247108204320607285918568\r\n'
+    b'Content-Disposition: form-data; name="uploadid"\r\n\r\n'
+    b'00l33t0174873295\r\n'
+    b'-----------------------------1574247108204320607285918568\r\n'
+    b'Content-Disposition: form-data; name="file"; filename="test.txt"\r\n'
+    b'Content-Type: text/plain\r\n\r\n'
+    b'Hello, world!\n'
+    b'\r\n'
+    b'-----------------------------1574247108204320607285918568--\r\n'
+)
+EXAMPLE2_PART_COUNT = 5
+
+EXAMPLE3 = (
+    b'--BOUNDARY\r\n'
+    b'Content-Disposition: form-data; name="file"; filename="bytes"\r\n'
+    b'Content-Type: application/x-falcon\r\n\r\n' +
+    b'123456789abcdef\n' * 64 * 1024 * 2 +
+    b'\r\n'
+    b'--BOUNDARY\r\n'
+    b'Content-Disposition: form-data; name="empty"\r\n'
+    b'Content-Type: text/plain\r\n\r\n'
+    b'\r\n'
+    b'--BOUNDARY--\r\n'
+)
+
+LOREM_IPSUM = (
+    'Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod '
+    'tempor incididunt ut labore et dolore magna aliqua. Dolor sed viverra '
+    'ipsum nunc aliquet bibendum enim. In massa tempor nec feugiat. Nunc '
+    'aliquet bibendum enim facilisis gravida. Nisl nunc mi ipsum faucibus '
+    'vitae aliquet nec ullamcorper. Amet luctus venenatis lectus magna '
+    'fringilla. Volutpat maecenas volutpat blandit aliquam etiam erat velit '
+    'scelerisque in. Egestas egestas fringilla phasellus faucibus scelerisque '
+    'eleifend. Sagittis orci a scelerisque purus semper eget duis. Nulla '
+    'pharetra diam sit amet nisl suscipit. Sed adipiscing diam donec '
+    'adipiscing tristique risus nec feugiat in. Fusce ut placerat orci nulla. '
+    'Pharetra vel turpis nunc eget lorem dolor. Tristique senectus et netus '
+    'et malesuada.\n'
+).encode()
+
+EXAMPLE4 = (
+    b'--boundary\r\n'
+    b'Content-Disposition: form-data; name="lorem1"; filename="bytes1"\r\n'
+    b'Content-Type: text/plain\r\n\r\n' +
+    LOREM_IPSUM +
+    b'\r\n'
+    b'--boundary\r\n'
+    b'Content-Disposition: form-data; name="empty"\r\n'
+    b'Content-Type: text/plain\r\n\r\n'
+    b'\r\n'
+    b'--boundary\r\n'
+    b'Content-Disposition: form-data; name="lorem2"; filename="bytes1"\r\n'
+    b'Content-Type: text/plain\r\n\r\n' +
+    LOREM_IPSUM +
+    b'\r\n'
+    b'--boundary--\r\n'
+)
+
+
+EXAMPLES = {
+    '5b11af82ab65407ba8cdccf37d2a9c4f': EXAMPLE1,
+    '---------------------------1574247108204320607285918568': EXAMPLE2,
+    'BOUNDARY': EXAMPLE3,
+    'boundary': EXAMPLE4,
+}
+
+
+# APPSEC: This SHA256 hash is safe against preimage attacks.
+HASH_BOUNDARY = (
+    'fbeff51e0f5630958701f4941aec5595addcb3ee1b70468c8bd4be920304c184')
+
+
+@pytest.mark.parametrize('boundary', list(EXAMPLES))
+def test_parse(boundary):
+    handler = media.MultipartFormHandler()
+    example = EXAMPLES[boundary]
+    form = handler.deserialize(
+        io.BytesIO(example),
+        'multipart/form-data; boundary=' + boundary,
+        len(example))
+
+    for part in form:
+        output = io.BytesIO()
+        part.stream.pipe(output)
+        assert isinstance(output.getvalue(), bytes)
+
+
+@pytest.mark.parametrize('buffer_size,chunk_size', list(itertools.product(
+    (
+        32,
+        64,
+        128,
+        256,
+    ),
+    (
+        7,
+        8,
+        9,
+        10,
+        32,
+        64,
+        128,
+        256,
+    ),
+)))
+def test_parsing_correctness(buffer_size, chunk_size):
+    example = EXAMPLES['boundary']
+    handler = media.MultipartFormHandler()
+    stream = BufferedReader(io.BytesIO(example).read, len(example),
+                            buffer_size)
+    form = handler.deserialize(
+        stream, 'multipart/form-data; boundary=boundary', len(example))
+
+    for part in form:
+        if part.name in ('lorem1', 'lorem2'):
+            part_stream = part.stream
+            result = []
+            while True:
+                chunk = part_stream.read(chunk_size)
+                if not chunk:
+                    break
+                result.append(chunk)
+
+            assert b''.join(result) == LOREM_IPSUM
+
+
+def test_missing_boundary():
+    handler = media.MultipartFormHandler()
+
+    with pytest.raises(falcon.HTTPInvalidHeader):
+        handler.deserialize(io.BytesIO(), 'multipart/form-data', 0)
+
+    with pytest.raises(falcon.HTTPInvalidHeader):
+        handler.deserialize(io.BytesIO(), 'multipart/form-data; boundary=', 0)
+
+    overlong = '-' * 71
+    content_type = 'multipart/form-data; boundary=' + overlong
+    with pytest.raises(falcon.HTTPInvalidHeader):
+        handler.deserialize(io.BytesIO(), content_type, 0)
+
+
+def test_empty_input():
+    handler = media.MultipartFormHandler()
+    form = handler.deserialize(
+        io.BytesIO(), 'multipart/form-data; boundary=404', 0)
+    with pytest.raises(falcon.HTTPBadRequest):
+        for part in form:
+            pass
+
+
+def test_serialize():
+    handler = media.MultipartFormHandler()
+    with pytest.raises(NotImplementedError):
+        handler.serialize({'key': 'value'}, 'multipart/form-data')
+
+
+def test_content_transfer_encoding_header():
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="file"; filename="bytes"\r\n'
+        b'Content-Transfer-Encoding: Base64'
+        b'Content-Type: application/x-falcon\r\n\r\n'
+        b'UGVyZWdyaW5lIEZhbGNvbiADLgA='
+        b'\r\n'
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="empty"\r\n'
+        b'Content-Type: text/plain\r\n\r\n'
+        b'\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    handler = media.MultipartFormHandler()
+    form = handler.deserialize(
+        io.BytesIO(data), 'multipart/form-data; boundary=BOUNDARY', len(data))
+    with pytest.raises(falcon.HTTPBadRequest):
+        for part in form:
+            pass
+
+
+def test_unsupported_charset():
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="text"\r\n'
+        b'Content-Type: text/plain; charset=pecyn\r\n\r\n'
+        b'AAHEHlRoZSBGYWxjb24gV2ViIEZyYW1ld29yaywgMjAxOQ=='
+        b'\r\n'
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="empty"\r\n'
+        b'Content-Type: text/plain\r\n\r\n'
+        b'\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    handler = media.MultipartFormHandler()
+    form = handler.deserialize(
+        io.BytesIO(data), 'multipart/form-data; boundary=BOUNDARY', len(data))
+    with pytest.raises(falcon.HTTPBadRequest):
+        for part in form:
+            part.text
+
+
+@pytest.mark.parametrize('charset,data', [
+    ('utf-8', b'Impossible byte: \xff'),
+    ('utf-8', b'Overlong... \xfc\x83\xbf\xbf\xbf\xbf ... sequence'),
+    ('ascii', b'\x80\x80\x80'),
+    ('pecyn', b'AAHEHlRoZSBGYWxjb24gV2ViIEZyYW1ld29yaywgMjAxOQ=='),
+])
+def test_invalid_text_or_charset(charset, data):
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="text"\r\n'
+        b'Content-Type: text/plain; ' +
+        'charset={}\r\n\r\n'.format(charset).encode() +
+        data +
+        b'\r\n'
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="empty"\r\n'
+        b'Content-Type: text/plain\r\n\r\n'
+        b'\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    handler = media.MultipartFormHandler()
+
+    form = handler.deserialize(
+        io.BytesIO(data), 'multipart/form-data; boundary=BOUNDARY', len(data))
+    with pytest.raises(falcon.HTTPBadRequest):
+        for part in form:
+            part.text
+
+
+def test_unknown_header():
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="empty"\r\n'
+        b'Content-Coolness: fair\r\n'
+        b'Content-Type: text/plain\r\n\r\n'
+        b'\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    handler = media.MultipartFormHandler()
+    form = handler.deserialize(
+        io.BytesIO(data), 'multipart/form-data; boundary=BOUNDARY', len(data))
+
+    for part in form:
+        assert part.data == b''
+
+
+def test_from_buffered_stream():
+    data = (
+        b'--BOUNDARY\r\n'
+        b'Content-Disposition: form-data; name="empty"\r\n'
+        b'Content-Coolness: fair\r\n'
+        b'Content-Type: text/plain\r\n\r\n'
+        b'\r\n'
+        b'--BOUNDARY--\r\n'
+    )
+
+    handler = media.MultipartFormHandler()
+    stream = BufferedReader(io.BytesIO(data).read, len(data))
+    form = handler.deserialize(
+        stream, 'multipart/form-data; boundary=BOUNDARY', len(data))
+
+    for part in form:
+        assert part.data == b''
+
+
+def test_body_part_media():
+    handler = media.MultipartFormHandler()
+
+    content_type = ('multipart/form-data; boundary=' +
+                    '5b11af82ab65407ba8cdccf37d2a9c4f')
+    form = handler.deserialize(
+        io.BytesIO(EXAMPLE1), content_type, len(EXAMPLE1))
+
+    expected = {'debug': True, 'message': 'Hello, world!', 'score': 7}
+
+    for part in form:
+        if part.content_type == 'application/json':
+            assert part.media == part.media == expected
+
+
+def test_body_part_properties():
+    handler = media.MultipartFormHandler()
+
+    content_type = ('multipart/form-data; boundary=' +
+                    '5b11af82ab65407ba8cdccf37d2a9c4f')
+    form = handler.deserialize(
+        io.BytesIO(EXAMPLE1), content_type, len(EXAMPLE1))
+
+    for part in form:
+        if part.content_type == 'application/json':
+            assert part.name == part.name == 'document'
+        elif part.name == 'file1':
+            assert part.filename == part.filename == 'test.txt'
+            assert part.secure_filename == part.filename
+
+
+def test_filename_star():
+    # NOTE(vytas): Generated by requests_toolbelt 0.9.1 on Py2
+    #   (interestingly, one gets a "normal" filename on Py3.7).
+
+    data = (
+        b'--a0d738bcdb30449eb0d13f4b72c2897e\r\n'
+        b'Content-Disposition: form-data; name="file"; '
+        b"filename*=utf-8''%E2%AC%85%20Arrow.txt\r\n\r\n"
+        b'A unicode arrow in the filename.\r\n'
+        b'--a0d738bcdb30449eb0d13f4b72c2897e--\r\n'
+    )
+
+    handler = media.MultipartFormHandler()
+    content_type = ('multipart/form-data; boundary=' +
+                    'a0d738bcdb30449eb0d13f4b72c2897e')
+    stream = BufferedReader(io.BytesIO(data).read, len(data))
+    form = handler.deserialize(stream, content_type, len(data))
+    for part in form:
+        assert part.filename == '⬅ Arrow.txt'
+        assert part.secure_filename == '__Arrow.txt'
+
+    data = data.replace(b'*=utf-8', b'*=esoteric')
+    stream = BufferedReader(io.BytesIO(data).read, len(data))
+    form = handler.deserialize(stream, content_type, len(data))
+    for part in form:
+        with pytest.raises(falcon.HTTPBadRequest):
+            part.filename
+
+
+def test_empty_filename():
+    data = (
+        b'--a0d738bcdb30449eb0d13f4b72c2897e\r\n'
+        b'Content-Disposition: form-data; name="file"; filename=\r\n\r\n'
+        b'An empty filename.\r\n'
+        b'--a0d738bcdb30449eb0d13f4b72c2897e--\r\n'
+    )
+
+    handler = media.MultipartFormHandler()
+    content_type = ('multipart/form-data; boundary=' +
+                    'a0d738bcdb30449eb0d13f4b72c2897e')
+    stream = BufferedReader(io.BytesIO(data).read, len(data))
+    form = handler.deserialize(stream, content_type, len(data))
+
+    for part in form:
+        assert part.filename == ''
+        with pytest.raises(falcon.HTTPBadRequest):
+            part.secure_filename
+
+
+@pytest.mark.parametrize('max_headers_size', [64, 140, 141, 142, 256, 1024])
+def test_headers_edge_cases(max_headers_size):
+    data = (
+        b'--a0d738bcdb30449eb0d13f4b72c2897e\r\n'
+        b'X-Falcon: Peregrine\r\n'
+        b'Content-Type: application/vnd.oasis.opendocument.text\r\n'
+        b'Junk\r\n'
+        b'Content-Disposition: form-data; name="file"; filename=hd.txt\r\n\r\n'
+        b'No, it is not an ODT document...\r\n'
+        b'--a0d738bcdb30449eb0d13f4b72c2897e--\r\n'
+    )
+
+    handler = media.MultipartFormHandler()
+    handler.parse_options.max_body_part_headers_size = max_headers_size
+
+    content_type = ('multipart/form-data; boundary=' +
+                    'a0d738bcdb30449eb0d13f4b72c2897e')
+    stream = BufferedReader(io.BytesIO(data).read, len(data))
+    form = handler.deserialize(stream, content_type, len(data))
+
+    if max_headers_size < 142:
+        with pytest.raises(falcon.HTTPBadRequest):
+            list(form)
+    else:
+        assert len(list(form)) == 1
+
+
+class MultipartAnalyzer:
+    def on_post(self, req, resp):
+        values = []
+        for part in req.media:
+            values.append({
+                'content_type': part.content_type,
+                'data': part.data.decode(),
+                'filename': part.filename,
+                'name': part.name,
+                'text': part.text,
+            })
+
+        resp.media = values
+
+
+@pytest.fixture
+def client():
+    app = falcon.App()
+    app.add_route('/media', MultipartAnalyzer())
+
+    return testing.TestClient(app)
+
+
+@pytest.fixture
+def custom_client():
+    def _factory(options):
+        multipart_handler = media.MultipartFormHandler()
+        for key, value in options.items():
+            setattr(multipart_handler.parse_options, key, value)
+        handlers = media.Handlers({
+            falcon.MEDIA_JSON: media.JSONHandler(),
+            falcon.MEDIA_MULTIPART: multipart_handler,
+        })
+
+        app = falcon.App()
+        app.req_options.media_handlers = handlers
+        app.add_route('/media', MultipartAnalyzer())
+
+        return testing.TestClient(app)
+
+    return _factory
+
+
+def test_upload_multipart(client):
+    resp = client.simulate_post(
+        '/media',
+        headers={
+            'Content-Type':
+            'multipart/form-data; boundary=5b11af82ab65407ba8cdccf37d2a9c4f',
+        },
+        body=EXAMPLE1)
+
+    assert resp.status_code == 200
+    assert resp.json == [
+        {
+            'content_type': 'text/plain',
+            'data': 'world',
+            'filename': None,
+            'name': 'hello',
+            'text': 'world',
+        },
+        {
+            'content_type': 'application/json',
+            'data': '{"debug": true, "message": "Hello, world!", "score": 7}',
+            'filename': None,
+            'name': 'document',
+            'text': None,
+        },
+        {
+            'content_type': 'text/plain',
+            'data': 'Hello, world!\n',
+            'filename': 'test.txt',
+            'name': 'file1',
+            'text': 'Hello, world!\n',
+        },
+    ]
+
+
+@pytest.mark.parametrize('truncated_by', [1, 2, 3, 4])
+def test_truncated_form(client, truncated_by):
+    resp = client.simulate_post(
+        '/media',
+        headers={
+            'Content-Type':
+            'multipart/form-data; boundary=5b11af82ab65407ba8cdccf37d2a9c4f',
+        },
+        body=EXAMPLE1[:-truncated_by])
+
+    assert resp.status_code == 400
+    assert resp.json == {
+        'description': 'unexpected form structure',
+        'title': 'Malformed multipart/form-data request media',
+    }
+
+
+def test_unexected_form_structure(client):
+    resp1 = client.simulate_post(
+        '/media',
+        headers={
+            'Content-Type':
+            'multipart/form-data; boundary=5b11af82ab65407ba8cdccf37d2a9c4f',
+        },
+        body=EXAMPLE1[:-2] + b'--\r\n')
+
+    assert resp1.status_code == 400
+    assert resp1.json == {
+        'description': 'unexpected form structure',
+        'title': 'Malformed multipart/form-data request media',
+    }
+
+    resp2 = client.simulate_post(
+        '/media',
+        headers={
+            'Content-Type':
+            'multipart/form-data; boundary=5b11af82ab65407ba8cdccf37d2a9c4f',
+        },
+        body=EXAMPLE1[:-4] + b'**\r\n')
+
+    assert resp2.status_code == 400
+    assert resp2.json == {
+        'description': 'unexpected form structure',
+        'title': 'Malformed multipart/form-data request media',
+    }
+
+
+def test_data_too_large(client):
+    resp = client.simulate_post(
+        '/media',
+        headers={
+            'Content-Type':
+            'multipart/form-data; boundary=BOUNDARY',
+        },
+        body=EXAMPLE3)
+
+    assert resp.status_code == 400
+    assert resp.json == {
+        'description': 'body part is too large',
+        'title': 'Malformed multipart/form-data request media',
+    }
+
+
+@pytest.mark.parametrize('max_body_part_count', list(range(7)) + [100, 1000])
+def test_too_many_body_parts(custom_client, max_body_part_count):
+    client = custom_client({'max_body_part_count': max_body_part_count})
+    boundary = '---------------------------1574247108204320607285918568'
+    resp = client.simulate_post(
+        '/media',
+        headers={
+            'Content-Type':
+            'multipart/form-data; boundary=' + boundary,
+        },
+        body=EXAMPLE2)
+
+    if 0 < max_body_part_count < EXAMPLE2_PART_COUNT:
+        assert resp.status_code == 400
+        assert resp.json == {
+            'description': 'maximum number of form body parts exceeded',
+            'title': 'Malformed multipart/form-data request media',
+        }
+    else:
+        assert resp.status_code == 200
+        assert len(resp.json) == EXAMPLE2_PART_COUNT
+
+
+def test_random_form(client):
+    part_data = [os.urandom(random.randint(0, 2**18)) for _ in range(64)]
+    form_data = b''.join(
+        '--{}\r\n'.format(HASH_BOUNDARY).encode() +
+        'Content-Disposition: form-data; name="p{}"\r\n'.format(i).encode() +
+        b'Content-Type: application/x-falcon-urandom\r\n\r\n' +
+        part_data[i] +
+        b'\r\n'
+        for i in range(64)
+    ) + '--{}--\r\n'.format(HASH_BOUNDARY).encode()
+
+    handler = media.MultipartFormHandler()
+    content_type = ('multipart/form-data; boundary=' + HASH_BOUNDARY)
+    form = handler.deserialize(
+        io.BytesIO(form_data), content_type, len(form_data))
+
+    for index, part in enumerate(form):
+        assert part.content_type == 'application/x-falcon-urandom'
+        assert part.data == part_data[index]
+
+
+def test_invalid_random_form(client):
+    length = random.randint(2**20, 2**21)
+    resp = client.simulate_post(
+        '/media',
+        headers={
+            'Content-Type':
+            'multipart/form-data; boundary=' + HASH_BOUNDARY,
+        },
+        body=os.urandom(length))
+
+    assert resp.status_code == 400
diff --git a/tests/test_media_urlencoded.py b/tests/test_media_urlencoded.py
index 5ca61c2..c312346 100644
--- a/tests/test_media_urlencoded.py
+++ b/tests/test_media_urlencoded.py
@@ -19,7 +19,7 @@ def test_deserialize_invalid_unicode():
     handler = media.URLEncodedFormHandler()
     stream = io.BytesIO('spade=♠'.encode())
     with pytest.raises(UnicodeDecodeError):
-        print(handler.deserialize(stream, falcon.MEDIA_URLENCODED, 9))
+        handler.deserialize(stream, falcon.MEDIA_URLENCODED, 9)
 
 
 @pytest.mark.parametrize('data,expected', [
diff --git a/tests/test_request_media.py b/tests/test_request_media.py
index e3c5e53..27ba1f7 100644
--- a/tests/test_request_media.py
+++ b/tests/test_request_media.py
@@ -177,6 +177,8 @@ class NopeHandler(media.BaseHandler):
     def deserialize(self, *args, **kwargs):
         pass
 
+    exhaust_stream = True
+
 
 def test_complete_consumption(asgi):
     client = create_client(asgi, {
diff --git a/tests/test_utils.py b/tests/test_utils.py
index a9d63be..c813257 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -482,6 +482,22 @@ class TestFalconUtils:
         assert not weak_67ab43_one.strong_compare(weak_67aB43)
         assert not weak_67aB43.strong_compare(weak_67ab43_one)
 
+    @pytest.mark.parametrize('filename,expected', [
+        ('.', '_'),
+        ('..', '_.'),
+        ('hello.txt', 'hello.txt'),
+        ('Ąžuolai žaliuos.jpeg', 'A_z_uolai_z_aliuos.jpeg'),
+        ('/etc/shadow', '_etc_shadow'),
+        ('. ⬅ a dot', '____a_dot'),
+        ('C:\\Windows\\kernel32.dll', 'C__Windows_kernel32.dll'),
+    ])
+    def test_secure_filename(self, filename, expected):
+        assert misc.secure_filename(filename) == expected
+
+    def test_secure_filename_empty_value(self):
+        with pytest.raises(ValueError):
+            misc.secure_filename('')
+
 
 @pytest.mark.parametrize(
     'protocol,method',