Support for streaming request body handling

tornado/httpserver.py

@@ -240,10 +240,7 @@ def _on_headers(self, data):
                 content_length = int(content_length)
                 if content_length > self.stream.max_buffer_size:
                     raise _BadRequestException("Content-Length too long")
-                if headers.get("Expect") == "100-continue":
-                    self.stream.write(b("HTTP/1.1 100 (Continue)\r\n\r\n"))
-                self.stream.read_bytes(content_length, self._on_request_body)
-                return
+                self._request.content_length = content_length
 
             self.request_callback(self._request)
         except _BadRequestException, e:
@@ -252,31 +249,6 @@ def _on_headers(self, data):
             self.stream.close()
             return
 
-    def _on_request_body(self, data):
-        self._request.body = data
-        content_type = self._request.headers.get("Content-Type", "")
-        if self._request.method in ("POST", "PUT"):
-            if content_type.startswith("application/x-www-form-urlencoded"):
-                arguments = parse_qs_bytes(native_str(self._request.body))
-                for name, values in arguments.iteritems():
-                    values = [v for v in values if v]
-                    if values:
-                        self._request.arguments.setdefault(name, []).extend(
-                            values)
-            elif content_type.startswith("multipart/form-data"):
-                fields = content_type.split(";")
-                for field in fields:
-                    k, sep, v = field.strip().partition("=")
-                    if k == "boundary" and v:
-                        httputil.parse_multipart_form_data(
-                            utf8(v), data,
-                            self._request.arguments,
-                            self._request.files)
-                        break
-                else:
-                    logging.warning("Invalid multipart/form-data")
-        self.request_callback(self._request)
-
 
 class HTTPRequest(object):
     """A single HTTP request.
@@ -392,7 +364,42 @@ def __init__(self, method, uri, version="HTTP/1.0", headers=None,
         for name, values in arguments.iteritems():
             values = [v for v in values if v]
             if values: self.arguments[name] = values
-
+
+    def request_continue(self):
+        '''Send a 100-Continue, telling the client to send the request body'''
+        if self.headers.get("Expect") == "100-continue":
+            self.connection.stream.write(b("HTTP/1.1 100 (Continue)\r\n\r\n"))
+
+    def _read_body(self, exec_req_cb):
+        self.request_continue()
+        self.connection.stream.read_bytes(self.content_length,
+            lambda data: self._on_request_body(data, exec_req_cb))
+
+    def _on_request_body(self, data, exec_req_cb):
+        self.body = data
+        content_type = self.headers.get("Content-Type", "")
+        if self.method in ("POST", "PUT"):
+            if content_type.startswith("application/x-www-form-urlencoded"):
+                arguments = parse_qs_bytes(native_str(self.body))
+                for name, values in arguments.iteritems():
+                    values = [v for v in values if v]
+                    if values:
+                        self.arguments.setdefault(name, []).extend(
+                            values)
+            elif content_type.startswith("multipart/form-data"):
+                fields = content_type.split(";")
+                for field in fields:
+                    k, sep, v = field.strip().partition("=")
+                    if k == "boundary" and v:
+                        httputil.parse_multipart_form_data(
+                            utf8(v), data,
+                            self.arguments,
+                            self.files)
+                        break
+                else:
+                    logging.warning("Invalid multipart/form-data")
+        exec_req_cb()
+
     def supports_http_1_1(self):
         """Returns True if this request supports HTTP/1.1 semantics"""
         return self.version == "HTTP/1.1"

tornado/web.py

@@ -966,9 +966,21 @@ def _execute(self, transforms, *args, **kwargs):
                 args = [self.decode_argument(arg) for arg in args]
                 kwargs = dict((k, self.decode_argument(v, name=k))
                               for (k,v) in kwargs.iteritems())
-                getattr(self, self.request.method.lower())(*args, **kwargs)
-                if self._auto_finish and not self._finished:
-                    self.finish()
+                # read and parse the request body, if not disabled
+                exec_req_cb = lambda: self._execute_request(*args, **kwargs)
+                if (getattr(self, '_read_body', True) and
+                    hasattr(self.request, 'content_length')):
+                    self.request._read_body(exec_req_cb)
+                else:
+                    exec_req_cb()
+        except Exception, e:
+            self._handle_request_exception(e)
+
+    def _execute_request(self, *args, **kwargs):
+        try:
+            getattr(self, self.request.method.lower())(*args, **kwargs)
+            if self._auto_finish and not self._finished:
+                self.finish()
         except Exception, e:
             self._handle_request_exception(e)
 
@@ -1026,6 +1038,56 @@ def _ui_method(self, method):
         return lambda *args, **kwargs: method(self, *args, **kwargs)
 
 
+def stream_body(cls):
+    """Wrap RequestHandler classes with this to prevent the
+    request body of PUT and POST request from being read and parsed
+    automatically.
+
+    If this decorator is given, the request body is not read and parsed
+    when a PUT or POST handler method is executed. It is up to the request
+    handler to read the body from the stream in the HTTP connection.
+
+    Using this decorator automatically implies the asynchronous decorator.
+
+    Without this decorator, the request body is automatically read and
+    parsed before a PUT or POST method is executed. ::
+
+       @web.stream_body
+       class StreamHandler(web.RequestHandler):
+
+           def put(self):
+               self.read_bytes = 0
+               self.request.request_continue()
+               self.read_chunks()
+
+           def read_chunks(self, chunk=''):
+               self.read_bytes += len(chunk)
+               chunk_length = min(10000,
+                   self.request.content_length - self.read_bytes)
+               if chunk_length > 0:
+                 self.request.connection.stream.read_bytes(
+                     chunk_length, self.read_chunks)
+               else:
+                   self.uploaded()
+
+           def uploaded(self):
+               self.write('Uploaded %d bytes' % self.read_bytes)
+               self.finish()
+
+    """
+    class StreamBody(cls):
+        def __init__(self, *args, **kwargs):
+            if args[0]._wsgi:
+                raise Exception("@stream_body is not supported for WSGI apps")
+            self._read_body = False
+            if hasattr(cls, 'post'):
+                cls.post = asynchronous(cls.post)
+            if hasattr(cls, 'put'):
+                cls.put = asynchronous(cls.put)
+            cls.__init__(self, *args, **kwargs)
+    return StreamBody
+
+
 def asynchronous(method):
     """Wrap request handler methods with this if they are asynchronous.