Skip to main content

Request & Response Handling

Comprehensive patterns for handling HTTP requests and responses in Tornado. Copy-paste ready code examples for all common scenarios.

Request Object​

Accessing Request Data​

import tornado.web
import tornado.escape

class RequestHandler(tornado.web.RequestHandler):
    def get(self):
        # Request method
        method = self.request.method  # GET, POST, PUT, DELETE, etc.

        # Request URI and path
        uri = self.request.uri         # /path?query=value
        path = self.request.path       # /path
        query = self.request.query     # query=value

        # Request headers
        headers = self.request.headers
        user_agent = self.request.headers.get("User-Agent", "")
        content_type = self.request.headers.get("Content-Type", "")

        # Client information
        remote_ip = self.request.remote_ip
        host = self.request.host

        # Request body
        body = self.request.body  # Raw bytes

        # Protocol information
        protocol = self.request.protocol  # HTTP/1.1, HTTP/2.0
        version = self.request.version    # HTTP version

        self.write({
            "method": method,
            "uri": uri,
            "path": path,
            "query": query,
            "remote_ip": remote_ip,
            "host": host,
            "user_agent": user_agent
        })

Request Arguments​

URL Arguments (from URL patterns)​

class UserHandler(tornado.web.RequestHandler):
    def get(self, user_id):
        # URL pattern: r"/user/([0-9]+)"
        self.write(f"User ID: {user_id}")

    def post(self, user_id):
        # Multiple URL arguments
        # URL pattern: r"/user/([0-9]+)/post/([0-9]+)"
        post_id = self.path_args[1]  # Second argument
        self.write(f"User {user_id}, Post {post_id}")

Query String Arguments​

class SearchHandler(tornado.web.RequestHandler):
    def get(self):
        # Single argument
        query = self.get_argument("q", default="")

        # Multiple arguments with same name
        tags = self.get_arguments("tag")  # Returns list

        # Arguments with type conversion
        page = int(self.get_argument("page", default="1"))
        limit = int(self.get_argument("limit", default="10"))

        # Check if argument exists
        if self.get_argument("format", default=None):
            format_type = self.get_argument("format")

        self.write({
            "query": query,
            "tags": tags,
            "page": page,
            "limit": limit
        })

Body Arguments (from form data)​

class FormHandler(tornado.web.RequestHandler):
    def post(self):
        # Single body argument
        name = self.get_body_argument("name")

        # Multiple body arguments
        emails = self.get_body_arguments("email")

        # Optional body argument
        message = self.get_body_argument("message", default="")

        # Check if body argument exists
        if self.get_body_argument("newsletter", default=None):
            newsletter = True
        else:
            newsletter = False

        self.write({
            "name": name,
            "emails": emails,
            "message": message,
            "newsletter": newsletter
        })

JSON Request Data​

import tornado.escape
import json

class JSONHandler(tornado.web.RequestHandler):
    def post(self):
        try:
            # Parse JSON from request body
            json_data = tornado.escape.json_decode(self.request.body)

            # Access JSON fields
            name = json_data.get("name")
            age = json_data.get("age")
            email = json_data.get("email")

            # Validate required fields
            if not name or not email:
                self.set_status(400)
                self.write({"error": "Name and email are required"})
                return

            # Process data
            response = {
                "status": "success",
                "data": {
                    "name": name,
                    "age": age,
                    "email": email
                }
            }

            self.write(json.dumps(response))

        except ValueError as e:
            self.set_status(400)
            self.write({"error": "Invalid JSON format"})
        except Exception as e:
            self.set_status(500)
            self.write({"error": "Internal server error"})

File Uploads​

import os
import uuid

class FileUploadHandler(tornado.web.RequestHandler):
    def post(self):
        # Single file upload
        if "file" not in self.request.files:
            self.set_status(400)
            self.write({"error": "No file uploaded"})
            return

        file_info = self.request.files["file"][0]
        filename = file_info["filename"]
        content_type = file_info["content_type"]
        body = file_info["body"]

        # Generate unique filename
        file_extension = os.path.splitext(filename)[1]
        unique_filename = str(uuid.uuid4()) + file_extension

        # Save file
        upload_path = os.path.join("uploads", unique_filename)
        with open(upload_path, "wb") as f:
            f.write(body)

        self.write({
            "status": "success",
            "filename": filename,
            "saved_as": unique_filename,
            "size": len(body),
            "content_type": content_type
        })

    def put(self):
        # Multiple file uploads
        uploaded_files = []

        for field_name in self.request.files:
            for file_info in self.request.files[field_name]:
                filename = file_info["filename"]
                body = file_info["body"]

                # Save each file
                file_extension = os.path.splitext(filename)[1]
                unique_filename = str(uuid.uuid4()) + file_extension
                upload_path = os.path.join("uploads", unique_filename)

                with open(upload_path, "wb") as f:
                    f.write(body)

                uploaded_files.append({
                    "original_name": filename,
                    "saved_as": unique_filename,
                    "size": len(body)
                })

        self.write({
            "status": "success",
            "files": uploaded_files
        })

Response Handling​

Basic Response Methods​

class ResponseHandler(tornado.web.RequestHandler):
    def get(self):
        # Write string response
        self.write("Hello, World!")

        # Write dictionary (automatically converted to JSON)
        self.write({"message": "Hello, JSON!"})

        # Write with explicit finish
        self.write("Response sent")
        self.finish()  # Explicitly finish response

    def post(self):
        # Write and flush
        self.write("Processing...")
        self.flush()  # Send partial response

        # Continue processing
        import time
        time.sleep(2)

        self.write("Done!")
        self.finish()

Setting Response Headers​

class HeaderHandler(tornado.web.RequestHandler):
    def set_default_headers(self):
        # Set headers for all requests
        self.set_header("Server", "Tornado/6.4")
        self.set_header("X-Custom-Header", "Custom Value")

    def get(self):
        # Set specific headers
        self.set_header("Content-Type", "application/json")
        self.set_header("Cache-Control", "no-cache")
        self.set_header("X-Request-ID", str(uuid.uuid4()))

        # Set multiple headers
        headers = {
            "X-API-Version": "1.0",
            "X-Rate-Limit": "1000",
            "X-Rate-Remaining": "999"
        }
        for key, value in headers.items():
            self.set_header(key, value)

        self.write({"status": "success"})

    def post(self):
        # CORS headers
        self.set_header("Access-Control-Allow-Origin", "*")
        self.set_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS")
        self.set_header("Access-Control-Allow-Headers", "Content-Type")

        self.write({"message": "CORS enabled"})

Response Status Codes​

class StatusHandler(tornado.web.RequestHandler):
    def get(self):
        # Success responses
        self.set_status(200)  # OK (default)
        self.write({"status": "success"})

    def post(self):
        # Created
        self.set_status(201)
        self.write({"status": "created"})

    def put(self):
        # No content
        self.set_status(204)
        self.finish()  # No response body

    def delete(self):
        # Custom status with reason
        self.set_status(410, "Gone")
        self.write({"error": "Resource no longer available"})

class ErrorHandler(tornado.web.RequestHandler):
    def get(self):
        # Client errors
        self.set_status(400)  # Bad Request
        self.write({"error": "Bad request"})

    def post(self):
        # Unauthorized
        self.set_status(401)
        self.write({"error": "Unauthorized"})

    def put(self):
        # Forbidden
        self.set_status(403)
        self.write({"error": "Forbidden"})

    def delete(self):
        # Not found
        self.set_status(404)
        self.write({"error": "Not found"})

    def patch(self):
        # Server error
        self.set_status(500)
        self.write({"error": "Internal server error"})

JSON Responses​

import json
import datetime

class JSONResponseHandler(tornado.web.RequestHandler):
    def set_default_headers(self):
        self.set_header("Content-Type", "application/json")

    def get(self):
        # Simple JSON response
        data = {
            "message": "Hello, JSON!",
            "timestamp": datetime.datetime.now().isoformat(),
            "data": ["item1", "item2", "item3"]
        }
        self.write(json.dumps(data))

    def post(self):
        # JSON with custom encoder
        class CustomJSONEncoder(json.JSONEncoder):
            def default(self, obj):
                if isinstance(obj, datetime.datetime):
                    return obj.isoformat()
                return super().default(obj)

        data = {
            "created_at": datetime.datetime.now(),
            "user_id": 123,
            "status": "active"
        }

        json_string = json.dumps(data, cls=CustomJSONEncoder)
        self.write(json_string)

    def put(self):
        # Pretty printed JSON
        data = {
            "users": [
                {"id": 1, "name": "John"},
                {"id": 2, "name": "Jane"}
            ]
        }

        json_string = json.dumps(data, indent=2, sort_keys=True)
        self.write(json_string)

Redirects​

class RedirectHandler(tornado.web.RequestHandler):
    def get(self):
        # Permanent redirect (301)
        self.redirect("https://example.com/new-location", permanent=True)

    def post(self):
        # Temporary redirect (302)
        self.redirect("/dashboard")

    def put(self):
        # Redirect with status code
        self.redirect("/login", status=303)

    def delete(self):
        # Conditional redirect
        if not self.current_user:
            self.redirect("/login")
            return

        # Continue with deletion
        self.write({"status": "deleted"})

Streaming Responses​

import asyncio

class StreamHandler(tornado.web.RequestHandler):
    async def get(self):
        # Set headers for streaming
        self.set_header("Content-Type", "text/plain")
        self.set_header("Cache-Control", "no-cache")

        # Stream data
        for i in range(10):
            self.write(f"Chunk {i}\n")
            self.flush()
            await asyncio.sleep(1)

        self.finish()

    async def post(self):
        # Server-sent events
        self.set_header("Content-Type", "text/event-stream")
        self.set_header("Cache-Control", "no-cache")
        self.set_header("Connection", "keep-alive")

        for i in range(5):
            self.write(f"data: Event {i}\n\n")
            self.flush()
            await asyncio.sleep(2)

        self.finish()

Cookies​

Setting Cookies​

import datetime

class CookieHandler(tornado.web.RequestHandler):
    def get(self):
        # Simple cookie
        self.set_cookie("username", "john_doe")

        # Cookie with expiration
        expires = datetime.datetime.now() + datetime.timedelta(days=30)
        self.set_cookie("session", "abc123", expires=expires)

        # Secure cookie
        self.set_secure_cookie("user_id", "123")

        # Cookie with all options
        self.set_cookie(
            name="preferences",
            value="dark_theme",
            expires=expires,
            domain=".example.com",
            path="/",
            secure=True,
            httponly=True,
            samesite="Lax"
        )

        self.write({"status": "cookies set"})

    def post(self):
        # Reading cookies
        username = self.get_cookie("username", default="anonymous")
        session = self.get_cookie("session")
        user_id = self.get_secure_cookie("user_id")

        # Decode secure cookie
        if user_id:
            user_id = user_id.decode('utf-8')

        self.write({
            "username": username,
            "session": session,
            "user_id": user_id
        })

    def delete(self):
        # Clear cookies
        self.clear_cookie("username")
        self.clear_cookie("session")
        self.clear_cookie("user_id")

        self.write({"status": "cookies cleared"})
import tornado.web
import base64
import os

class SecureCookieApp(tornado.web.Application):
    def __init__(self):
        # Generate secure cookie secret
        cookie_secret = base64.b64encode(os.urandom(32)).decode('ascii')

        settings = {
            "cookie_secret": cookie_secret,
            "secure_cookies": True,  # Require HTTPS
        }

        handlers = [
            (r"/", SecureCookieHandler),
        ]

        super().__init__(handlers, **settings)

class SecureCookieHandler(tornado.web.RequestHandler):
    def get(self):
        # Set secure cookie
        self.set_secure_cookie("user_data", "sensitive_info")
        self.write({"status": "secure cookie set"})

    def post(self):
        # Read secure cookie
        user_data = self.get_secure_cookie("user_data")
        if user_data:
            user_data = user_data.decode('utf-8')

        self.write({"user_data": user_data})

Request Validation​

Input Validation​

import re
import tornado.web

class ValidationHandler(tornado.web.RequestHandler):
    def validate_email(self, email):
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return re.match(pattern, email) is not None

    def validate_phone(self, phone):
        pattern = r'^\+?1?[2-9]\d{2}[2-9]\d{2}\d{4}$'
        return re.match(pattern, phone) is not None

    def post(self):
        # Required fields
        name = self.get_body_argument("name", default="")
        email = self.get_body_argument("email", default="")
        phone = self.get_body_argument("phone", default="")

        errors = []

        # Validate required fields
        if not name.strip():
            errors.append("Name is required")

        if not email.strip():
            errors.append("Email is required")
        elif not self.validate_email(email):
            errors.append("Invalid email format")

        # Validate optional fields
        if phone and not self.validate_phone(phone):
            errors.append("Invalid phone number format")

        if errors:
            self.set_status(400)
            self.write({"errors": errors})
            return

        # Process valid data
        self.write({
            "status": "success",
            "data": {
                "name": name,
                "email": email,
                "phone": phone
            }
        })

Content Type Validation​

class ContentTypeHandler(tornado.web.RequestHandler):
    def post(self):
        content_type = self.request.headers.get("Content-Type", "")

        if content_type.startswith("application/json"):
            try:
                data = tornado.escape.json_decode(self.request.body)
                self.write({"status": "JSON received", "data": data})
            except ValueError:
                self.set_status(400)
                self.write({"error": "Invalid JSON"})

        elif content_type.startswith("application/x-www-form-urlencoded"):
            name = self.get_body_argument("name", default="")
            self.write({"status": "Form data received", "name": name})

        elif content_type.startswith("multipart/form-data"):
            if self.request.files:
                file_count = len(self.request.files)
                self.write({"status": "Files received", "count": file_count})
            else:
                self.write({"status": "No files uploaded"})

        else:
            self.set_status(415)
            self.write({"error": "Unsupported content type"})

Error Handling​

Custom Error Responses​

class ErrorResponseHandler(tornado.web.RequestHandler):
    def write_error(self, status_code, **kwargs):
        # Custom error response format
        error_data = {
            "error": {
                "code": status_code,
                "message": self._get_error_message(status_code)
            }
        }

        # Include exception details in debug mode
        if self.settings.get("serve_traceback"):
            import traceback
            error_data["error"]["traceback"] = traceback.format_exception(*kwargs["exc_info"])

        self.set_header("Content-Type", "application/json")
        self.write(json.dumps(error_data))

    def _get_error_message(self, status_code):
        messages = {
            400: "Bad Request",
            401: "Unauthorized",
            403: "Forbidden",
            404: "Not Found",
            405: "Method Not Allowed",
            500: "Internal Server Error"
        }
        return messages.get(status_code, "Unknown Error")

    def get(self):
        # Trigger various errors for testing
        error_type = self.get_argument("error", default="404")

        if error_type == "400":
            raise tornado.web.HTTPError(400, "Bad request example")
        elif error_type == "401":
            raise tornado.web.HTTPError(401, "Unauthorized access")
        elif error_type == "500":
            raise Exception("Internal server error example")
        else:
            raise tornado.web.HTTPError(404, "Page not found")

This comprehensive guide covers all the essential patterns for handling HTTP requests and responses in Tornado. The framework's strength lies in its ability to handle these operations asynchronously, which is covered in the async operations section.