Tornado-asynchronous web server

Tornado: Complete Guide to the Asynchronous Python Web Framework

What Is Tornado

Tornado is a powerful asynchronous web server and web framework for Python, specifically designed to handle a large number of concurrent connections. Created by the FriendFeed team (later acquired by Facebook), Tornado became one of the pioneers in asynchronous programming for Python.

The core philosophy of Tornado is to use non‑blocking I/O and an event‑driven architecture. This allows a single process to serve thousands of connections simultaneously, making it an ideal choice for building high‑traffic web applications, chats, real‑time systems, and API services.

Key Features and Benefits of Tornado

Asynchronous Architecture

Tornado uses an event‑driven programming model that enables handling many requests in a single thread. This significantly reduces context‑switch overhead and memory management costs.

Built‑in HTTP Server

Unlike many other frameworks, Tornado includes its own HTTP server, eliminating the need for additional components such as WSGI for deploying applications.

WebSocket Support

Tornado provides native WebSocket support, making it an excellent choice for creating interactive real‑time applications.

Compatibility with asyncio

Since version 5.0, Tornado is fully compatible with asyncio, allowing you to use it alongside other asynchronous Python libraries.

Flexible Routing System

Tornado offers a powerful URL routing system with support for regular expressions and parameters.

Built‑in Template Engine

The framework includes its own template engine for generating HTML pages with dynamic content.

Installation and Environment Setup

Installing Tornado

pip install tornado

To work with additional features you can install extra packages:

# MongoDB support
pip install motor

# PostgreSQL support
pip install asyncpg

# Redis support
pip install aioredis

Project Structure

Recommended Tornado project layout:

project/
├── app.py                 # Main application file
├── handlers/              # Request handlers
│   ├── __init__.py
│   ├── base.py
│   ├── auth.py
│   └── api.py
├── models/                # Data models
│   ├── __init__.py
│   └── user.py
├── templates/             # HTML templates
│   ├── base.html
│   ├── index.html
│   └── error.html
├── static/                # Static assets
│   ├── css/
│   ├── js/
│   └── images/
├── config/                # Configuration files
│   ├── __init__.py
│   └── settings.py
└── requirements.txt       # Project dependencies

Creating Your First Application

Basic Example

import tornado.ioloop
import tornado.web

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("Welcome to Tornado!")

class AboutHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("About page")

def make_app():
    return tornado.web.Application([
        (r"/", MainHandler),
        (r"/about", AboutHandler),
    ])

if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    print("Server started at http://localhost:8888")
    tornado.ioloop.IOLoop.current().start()

Application Configuration with Settings

import tornado.web
import tornado.ioloop
import os

class Application(tornado.web.Application):
    def __init__(self):
        handlers = [
            (r"/", MainHandler),
            (r"/api/users", UserHandler),
        ]
        
        settings = {
            "debug": True,
            "template_path": os.path.join(os.path.dirname(__file__), "templates"),
            "static_path": os.path.join(os.path.dirname(__file__), "static"),
            "cookie_secret": "your-secret-key-here",
            "xsrf_cookies": True,
        }
        
        super().__init__(handlers, **settings)

Routing System

Fundamental Routing Principles

Tornado uses regular expressions to define routes. Each route is a tuple of a URL pattern and a handler class.

# Simple route
(r"/", MainHandler),

# Route with a parameter
(r"/user/([0-9]+)", UserHandler),

# Route with named groups
(r"/post/(?P<slug>[-\w]+)", PostHandler),

# Route with multiple parameters
(r"/category/([^/]+)/post/([0-9]+)", CategoryPostHandler),

Handling URL Parameters

class UserHandler(tornado.web.RequestHandler):
    def get(self, user_id):
        self.write(f"User profile ID: {user_id}")

class PostHandler(tornado.web.RequestHandler):
    def get(self, slug):
        self.write(f"Article: {slug}")

class CategoryPostHandler(tornado.web.RequestHandler):
    def get(self, category, post_id):
        self.write(f"Category: {category}, Post ID: {post_id}")

Request Handlers

Base RequestHandler Class

All Tornado handlers inherit from tornado.web.RequestHandler. This class provides methods for handling HTTP verbs:

class MyHandler(tornado.web.RequestHandler):
    def get(self):
        # Handle GET request
        self.write("GET request processed")
    
    def post(self):
        # Handle POST request
        name = self.get_argument("name")
        self.write(f"Hello, {name}!")
    
    def put(self):
        # Handle PUT request
        self.write("PUT request processed")
    
    def delete(self):
        # Handle DELETE request
        self.write("DELETE request processed")

Working with Query Parameters

class FormHandler(tornado.web.RequestHandler):
    def post(self):
        # Get required parameter
        name = self.get_argument("name")
        
        # Get parameter with default value
        age = self.get_argument("age", "18")
        
        # Get multiple values
        hobbies = self.get_arguments("hobbies")
        
        # Get all parameters
        all_args = self.request.arguments
        
        self.write(f"Name: {name}, Age: {age}, Hobbies: {hobbies}")

Working with Headers

class HeaderHandler(tornado.web.RequestHandler):
    def get(self):
        # Retrieve header
        user_agent = self.request.headers.get("User-Agent")
        
        # Set response header
        self.set_header("X-Custom-Header", "MyValue")
        
        # Set multiple headers
        self.set_header("Cache-Control", "no-cache")
        self.set_header("Content-Type", "application/json")
        
        self.write(f"Your User-Agent: {user_agent}")

Asynchronous Programming in Tornado

Async Basics

Tornado supports asynchronous programming using the async/await syntax:

import asyncio
import tornado.web
from tornado.httpclient import AsyncHTTPClient

class AsyncHandler(tornado.web.RequestHandler):
    async def get(self):
        # Asynchronous delay
        await asyncio.sleep(1)
        
        # Asynchronous HTTP request
        http_client = AsyncHTTPClient()
        response = await http_client.fetch("https://api.github.com/users/octocat")
        
        self.write(f"Response received: {response.code}")

Parallel Task Execution

import asyncio

class ParallelHandler(tornado.web.RequestHandler):
    async def get(self):
        # Launch several tasks in parallel
        tasks = [
            self.fetch_data("https://api.example1.com"),
            self.fetch_data("https://api.example2.com"),
            self.fetch_data("https://api.example3.com"),
        ]
        
        results = await asyncio.gather(*tasks)
        self.write(f"Received {len(results)} responses")
    
    async def fetch_data(self, url):
        http_client = AsyncHTTPClient()
        response = await http_client.fetch(url)
        return response.body

Template Handling

Template Basics

Tornado includes its own template engine that uses a Python‑like syntax:

<!-- templates/base.html -->
<!DOCTYPE html>
<html>
<head>
    <title>{% block title %}My Site{% end %}</title>
</head>
<body>
    <nav>
        <ul>
            <li><a href="/">Home</a></li>
            <li><a href="/about">About</a></li>
        </ul>
    </nav>
    
    <main>
        {% block content %}{% end %}
    </main>
    
    <footer>
        <p>&copy; 2024 My Site</p>
    </footer>
</body>
</html>

<!-- templates/index.html -->
{% extends "base.html" %}

{% block title %}Home - My Site{% end %}

{% block content %}
<h1>Welcome, {{ name }}!</h1>

{% if user %}
    <p>Logged in as {{ user.username }}</p>
{% else %}
    <p><a href="/login">Log in</a></p>
{% end %}

<ul>
{% for item in items %}
    <li>{{ item.title }} - {{ item.description }}</li>
{% end %}
</ul>
{% end %}

Using Templates in Handlers

class TemplateHandler(tornado.web.RequestHandler):
    def get(self):
        items = [
            {"title": "Product 1", "description": "Description for product 1"},
            {"title": "Product 2", "description": "Description for product 2"},
        ]
        
        self.render("index.html", 
                   name="User",
                   user={"username": "admin"},
                   items=items)

Custom Template Functions

def make_app():
    return tornado.web.Application([
        (r"/", MainHandler),
    ], 
    template_path="templates",
    ui_methods={
        "format_date": lambda handler, date: date.strftime("%d.%m.%Y"),
    },
    ui_modules={
        "UserCard": UserCardModule,
    })

Form and File Handling

Working with HTML Forms

<!-- templates/form.html -->
<form action="/submit" method="post">
    {% raw xsrf_form_html() %}
    <input type="text" name="name" placeholder="Name" required>
    <input type="email" name="email" placeholder="Email" required>
    <textarea name="message" placeholder="Message"></textarea>
    <button type="submit">Send</button>
</form>

class FormHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("form.html")
    
    def post(self):
        name = self.get_argument("name")
        email = self.get_argument("email")
        message = self.get_argument("message")
        
        # Process form data
        self.write(f"Thank you, {name}! Your message has been received.")

File Uploads

<!-- templates/upload.html -->
<form action="/upload" method="post" enctype="multipart/form-data">
    {% raw xsrf_form_html() %}
    <input type="file" name="file" accept=".jpg,.png,.gif" required>
    <button type="submit">Upload</button>
</form>

import os
import uuid

class UploadHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("upload.html")
    
    def post(self):
        if "file" not in self.request.files:
            self.write("No file selected")
            return
        
        file_info = self.request.files["file"][0]
        filename = file_info["filename"]
        file_body = file_info["body"]
        
        # Generate unique filename
        unique_filename = f"{uuid.uuid4()}_{filename}"
        
        # Save file
        upload_path = os.path.join("uploads", unique_filename)
        with open(upload_path, "wb") as f:
            f.write(file_body)
        
        self.write(f"File {filename} uploaded successfully")

WebSocket Connections

WebSocket Basics

import tornado.websocket
import json

class ChatWebSocket(tornado.websocket.WebSocketHandler):
    clients = set()
    
    def open(self):
        print(f"WebSocket connection opened: {self.request.remote_ip}")
        ChatWebSocket.clients.add(self)
        self.write_message({"type": "info", "message": "Welcome to the chat!"})
    
    def on_message(self, message):
        try:
            data = json.loads(message)
            
            if data["type"] == "chat":
                # Broadcast message to all connected clients
                for client in ChatWebSocket.clients:
                    client.write_message({
                        "type": "chat",
                        "user": data["user"],
                        "message": data["message"]
                    })
            
        except json.JSONDecodeError:
            self.write_message({"type": "error", "message": "Invalid data format"})
    
    def on_close(self):
        print(f"WebSocket connection closed: {self.request.remote_ip}")
        ChatWebSocket.clients.discard(self)
    
    def check_origin(self, origin):
        # Origin check – restrict in production
        return True

Client‑Side WebSocket Code

<!-- templates/chat.html -->
<div id="messages"></div>
<input type="text" id="messageInput" placeholder="Enter a message...">
<button onclick="sendMessage()">Send</button>

<script>
const ws = new WebSocket("ws://localhost:8888/ws");
const messages = document.getElementById('messages');
const messageInput = document.getElementById('messageInput');

ws.onmessage = function(event) {
    const data = JSON.parse(event.data);
    const messageElement = document.createElement('div');
    
    if (data.type === 'chat') {
        messageElement.textContent = '${data.user}: ${data.message}';
    } else if (data.type === 'info') {
        messageElement.textContent = data.message;
        messageElement.style.color = 'green';
    }
    
    messages.appendChild(messageElement);
    messages.scrollTop = messages.scrollHeight;
};

function sendMessage() {
    const message = messageInput.value.trim();
    if (message) {
        ws.send(JSON.stringify({
            type: 'chat',
            user: 'User',
            message: message
        }));
        messageInput.value = '';
    }
}

messageInput.addEventListener('keypress', function(e) {
    if (e.key === 'Enter') {
        sendMessage();
    }
});
</script>

Database Integration

Asynchronous MongoDB with Motor

import motor.motor_tornado
from bson import ObjectId

class DatabaseHandler:
    def __init__(self, db_url="mongodb://localhost:27017"):
        self.client = motor.motor_tornado.MotorClient(db_url)
        self.db = self.client.myapp
    
    async def create_user(self, user_data):
        result = await self.db.users.insert_one(user_data)
        return str(result.inserted_id)
    
    async def get_user(self, user_id):
        return await self.db.users.find_one({"_id": ObjectId(user_id)})
    
    async def get_all_users(self):
        cursor = self.db.users.find({})
        return await cursor.to_list(length=100)

# Usage in a handler
class UserHandler(tornado.web.RequestHandler):
    def initialize(self, db_handler):
        self.db = db_handler
    
    async def get(self, user_id=None):
        if user_id:
            user = await self.db.get_user(user_id)
            if user:
                user["_id"] = str(user["_id"])
                self.write(user)
            else:
                self.set_status(404)
                self.write({"error": "User not found"})
        else:
            users = await self.db.get_all_users()
            for user in users:
                user["_id"] = str(user["_id"])
            self.write({"users": users})
    
    async def post(self):
        user_data = {
            "name": self.get_argument("name"),
            "email": self.get_argument("email"),
            "created_at": datetime.utcnow()
        }
        user_id = await self.db.create_user(user_data)
        self.write({"id": user_id, "message": "User created"})

PostgreSQL with AsyncPG

import asyncpg
import asyncio

class PostgresHandler:
    def __init__(self, db_url="postgresql://user:password@localhost/dbname"):
        self.db_url = db_url
        self.pool = None
    
    async def init_pool(self):
        self.pool = await asyncpg.create_pool(self.db_url)
    
    async def create_user(self, name, email):
        async with self.pool.acquire() as conn:
            query = """
                INSERT INTO users (name, email, created_at) 
                VALUES ($1, $2, NOW()) 
                RETURNING id
            """
            return await conn.fetchval(query, name, email)
    
    async def get_user(self, user_id):
        async with self.pool.acquire() as conn:
            query = "SELECT * FROM users WHERE id = $1"
            return await conn.fetchrow(query, user_id)

Authentication and Authorization

Simple Authentication System

import hashlib
import tornado.web

class BaseHandler(tornado.web.RequestHandler):
    def get_current_user(self):
        user_id = self.get_secure_cookie("user_id")
        if user_id:
            return self.db.get_user(user_id.decode())
        return None

class LoginHandler(BaseHandler):
    def get(self):
        self.render("login.html", error=None)
    
    def post(self):
        username = self.get_argument("username")
        password = self.get_argument("password")
        
        # Verify user in DB
        user = self.authenticate_user(username, password)
        if user:
            self.set_secure_cookie("user_id", str(user["id"]))
            self.redirect("/dashboard")
        else:
            self.render("login.html", error="Invalid credentials")
    
    def authenticate_user(self, username, password):
        # Hash password
        password_hash = hashlib.sha256(password.encode()).hexdigest()
        # Find user in DB
        return self.db.find_user(username, password_hash)

class LogoutHandler(BaseHandler):
    def post(self):
        self.clear_cookie("user_id")
        self.redirect("/")

class DashboardHandler(BaseHandler):
    @tornado.web.authenticated
    def get(self):
        user = self.get_current_user()
        self.render("dashboard.html", user=user)

Authorization Decorators

import functools

def require_role(role):
    def decorator(method):
        @functools.wraps(method)
        def wrapper(self, *args, **kwargs):
            user = self.get_current_user()
            if not user or user.get("role") != role:
                self.set_status(403)
                self.write({"error": "Access denied"})
                return
            return method(self, *args, **kwargs)
        return wrapper
    return decorator

class AdminHandler(BaseHandler):
    @tornado.web.authenticated
    @require_role("admin")
    def get(self):
        self.write("Admin dashboard")

Middleware and Request Processing

prepare() Method

class BaseHandler(tornado.web.RequestHandler):
    def prepare(self):
        # Runs before each HTTP method
        self.set_header("X-Custom-Header", "MyApp")
        
        # Validate Content-Type for POST/PUT
        if self.request.method in ["POST", "PUT"]:
            content_type = self.request.headers.get("Content-Type", "")
            if not content_type.startswith("application/json"):
                self.set_status(400)
                self.write({"error": "Content-Type must be application/json"})
                self.finish()
                return
    
    def set_default_headers(self):
        # Runs for every request
        self.set_header("Access-Control-Allow-Origin", "*")
        self.set_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE")
        self.set_header("Access-Control-Allow-Headers", "Content-Type, Authorization")

Error Handling

class ErrorHandler(tornado.web.RequestHandler):
    def write_error(self, status_code, **kwargs):
        self.set_header("Content-Type", "application/json")
        
        if status_code == 404:
            self.write({"error": "Page not found"})
        elif status_code == 500:
            self.write({"error": "Internal server error"})
        else:
            self.write({"error": f"Error {status_code}"})
    
    def options(self):
        # Handle CORS preflight request
        self.set_status(204)
        self.finish()

Security

CSRF Protection

class SecureHandler(tornado.web.RequestHandler):
    def post(self):
        # Automatic CSRF token check
        # when xsrf_cookies=True in app settings
        name = self.get_argument("name")
        self.write(f"Data received: {name}")

Secure Cookies

class CookieHandler(tornado.web.RequestHandler):
    def post(self):
        # Set a secure cookie
        self.set_secure_cookie("user_session", "session_data", expires_days=7)
        
        # Retrieve secure cookie
        session_data = self.get_secure_cookie("user_session")
        
        self.write("Session set")

Data Validation

import re

class ValidationHandler(tornado.web.RequestHandler):
    def post(self):
        email = self.get_argument("email")
        
        # Validate email
        if not self.is_valid_email(email):
            self.set_status(400)
            self.write({"error": "Invalid email address"})
            return
        
        # Additional validation
        name = self.get_argument("name")
        if len(name) < 2:
            self.set_status(400)
            self.write({"error": "Name must be at least 2 characters"})
            return
        
        self.write({"message": "Data is valid"})
    
    def is_valid_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

Caching

In‑Memory Caching

import time
from functools import lru_cache

class CacheHandler(tornado.web.RequestHandler):
    _cache = {}
    
    async def get(self):
        cache_key = "expensive_data"
        
        # Check cache
        if cache_key in self._cache:
            data, timestamp = self._cache[cache_key]
            if time.time() - timestamp < 300:  # 5 minutes
                self.write({"data": data, "from_cache": True})
                return
        
        # Fetch data
        data = await self.get_expensive_data()
        
        # Store in cache
        self._cache[cache_key] = (data, time.time())
        
        self.write({"data": data, "from_cache": False})
    
    async def get_expensive_data(self):
        # Simulate expensive operation
        await asyncio.sleep(2)
        return {"result": "expensive computation"}

Caching with Redis

import aioredis
import json

class RedisHandler(tornado.web.RequestHandler):
    def initialize(self, redis_pool):
        self.redis = redis_pool
    
    async def get(self):
        cache_key = "user_data"
        
        # Try to fetch from Redis
        cached_data = await self.redis.get(cache_key)
        if cached_data:
            data = json.loads(cached_data)
            self.write({"data": data, "from_cache": True})
            return
        
        # Fetch from DB
        data = await self.get_user_data()
        
        # Store in Redis for 1 hour
        await self.redis.setex(cache_key, 3600, json.dumps(data))
        
        self.write({"data": data, "from_cache": False})

Testing Tornado Applications

Basic Testing

import unittest
import tornado.testing
import tornado.web

class TestHandler(tornado.web.RequestHandler):
    def get(self):
        self.write({"message": "Hello, World!"})

class TestApp(tornado.testing.AsyncHTTPTestCase):
    def get_app(self):
        return tornado.web.Application([
            (r"/", TestHandler),
        ])
    
    def test_homepage(self):
        response = self.fetch("/")
        self.assertEqual(response.code, 200)
        self.assertIn(b"Hello, World!", response.body)

if __name__ == "__main__":
    unittest.main()

WebSocket Testing

import tornado.testing
import tornado.websocket
import json

class EchoWebSocket(tornado.websocket.WebSocketHandler):
    def on_message(self, message):
        self.write_message(f"Echo: {message}")

class WebSocketTest(tornado.testing.AsyncHTTPTestCase):
    def get_app(self):
        return tornado.web.Application([
            (r"/ws", EchoWebSocket),
        ])
    
    @tornado.testing.gen_test
    def test_websocket(self):
        ws_url = "ws://localhost:" + str(self.get_http_port()) + "/ws"
        ws = yield tornado.websocket.websocket_connect(ws_url)
        
        ws.write_message("Hello")
        response = yield ws.read_message()
        
        self.assertEqual(response, "Echo: Hello")
        ws.close()

Production Deployment

Production Settings

import tornado.web
import tornado.httpserver
import tornado.ioloop
import tornado.options
import os

def create_app():
    settings = {
        "debug": False,
        "template_path": os.path.join(os.path.dirname(__file__), "templates"),
        "static_path": os.path.join(os.path.dirname(__file__), "static"),
        "cookie_secret": os.environ.get("COOKIE_SECRET"),
        "xsrf_cookies": True,
        "login_url": "/login",
    }
    
    return tornado.web.Application([
        (r"/", MainHandler),
        (r"/api/users", UserHandler),
        (r"/ws", WebSocketHandler),
    ], **settings)

def main():
    tornado.options.parse_command_line()
    
    app = create_app()
    
    # SSL configuration for HTTPS
    ssl_options = None
    if os.path.exists("cert.pem") and os.path.exists("key.pem"):
        ssl_options = {
            "certfile": "cert.pem",
            "keyfile": "key.pem",
        }
    
    server = tornado.httpserver.HTTPServer(app, ssl_options=ssl_options)
    server.listen(8888)
    
    print("Server running on port 8888")
    tornado.ioloop.IOLoop.current().start()

if __name__ == "__main__":
    main()

Docker Deployment

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8888

CMD ["python", "app.py"]

Nginx Configuration

upstream tornado {
    server 127.0.0.1:8888;
    server 127.0.0.1:8889;
}

server {
    listen 80;
    server_name example.com;
    
    location / {
        proxy_pass http://tornado;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
    
    location /static/ {
        alias /path/to/static/files/;
        expires 30d;
    }
    
    location /ws {
        proxy_pass http://tornado;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

Tornado Methods and Functions Reference

Useful Decorators and Utilities

Practical Usage Examples

REST API Server

import tornado.web
import tornado.ioloop
import json

class APIHandler(tornado.web.RequestHandler):
    def set_default_headers(self):
        self.set_header("Content-Type", "application/json")
        self.set_header("Access-Control-Allow-Origin", "*")
        self.set_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE")
        self.set_header("Access-Control-Allow-Headers", "Content-Type, Authorization")
    
    def options(self):
        self.set_status(204)
        self.finish()

class UsersHandler(APIHandler):
    def get(self):
        users = [
            {"id": 1, "name": "Ivan", "email": "ivan@example.com"},
            {"id": 2, "name": "Maria", "email": "maria@example.com"},
        ]
        self.write({"users": users})
    
    def post(self):
        try:
            data = json.loads(self.request.body)
            # Validate data
            if not data.get("name") or not data.get("email"):
                self.set_status(400)
                self.write({"error": "Name and email are required"})
                return
            
            # Create user
            user = {
                "id": 3,
                "name": data["name"],
                "email": data["email"]
            }
            
            self.set_status(201)
            self.write({"user": user, "message": "User created"})
            
        except json.JSONDecodeError:
            self.set_status(400)
            self.write({"error": "Invalid JSON"})

def make_app():
    return tornado.web.Application([
        (r"/api/users", UsersHandler),
    ])

if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    print("API server running at http://localhost:8888")
    tornado.ioloop.IOLoop.current().start()

Real‑Time Chat

import tornado.web
import tornado.websocket
import tornado.ioloop
import json
import uuid

class ChatHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("chat.html")

class ChatWebSocket(tornado.websocket.WebSocketHandler):
    connections = {}
    
    def open(self):
        self.id = str(uuid.uuid4())
        self.username = None
        ChatWebSocket.connections[self.id] = self
        print(f"New connection: {self.id}")
    
    def on_message(self, message):
        try:
            data = json.loads(message)
            
            if data["type"] == "join":
                self.username = data["username"]
                self.broadcast({
                    "type": "info",
                    "message": f"{self.username} joined the chat"
                })
                
            elif data["type"] == "message":
                if self.username:
                    self.broadcast({
                        "type": "message",
                        "username": self.username,
                        "message": data["message"],
                        "timestamp": data.get("timestamp")
                    })
                    
        except json.JSONDecodeError:
            self.write_message({
                "type": "error",
                "message": "Invalid message format"
            })
    
    def on_close(self):
        if self.id in ChatWebSocket.connections:
            del ChatWebSocket.connections[self.id]
            if self.username:
                self.broadcast({
                    "type": "info",
                    "message": f"{self.username} left the chat"
                })
        print(f"Connection closed: {self.id}")
    
    def broadcast(self, message):
        """Send a message to all connected clients"""
        for connection in ChatWebSocket.connections.values():
            try:
                connection.write_message(message)
            except:
                pass  # Ignore send errors
    
    def check_origin(self, origin):
        return True

def make_app():
    return tornado.web.Application([
        (r"/", ChatHandler),
        (r"/ws", ChatWebSocket),
    ], 
    template_path="templates",
    static_path="static",
    debug=True)

if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    print("Chat server running at http://localhost:8888")
    tornado.ioloop.IOLoop.current().start()

Performance Optimization

Connection Pools

import asyncio
import aioredis
import motor.motor_tornado

class ConnectionManager:
    def __init__(self):
        self.redis_pool = None
        self.mongo_client = None
    
    async def init_connections(self):
        # Redis connection pool
        self.redis_pool = aioredis.ConnectionPool.from_url(
            "redis://localhost:6379",
            max_connections=20
        )
        
        # MongoDB client
        self.mongo_client = motor.motor_tornado.MotorClient(
            "mongodb://localhost:27017",
            maxPoolSize=50
        )
    
    async def close_connections(self):
        if self.redis_pool:
            await self.redis_pool.disconnect()
        if self.mongo_client:
            self.mongo_client.close()

# Global connection manager
connection_manager = ConnectionManager()

class OptimizedHandler(tornado.web.RequestHandler):
    async def get(self):
        # Use connection pools
        redis = aioredis.Redis(connection_pool=connection_manager.redis_pool)
        
        # Parallel queries
        tasks = [
            redis.get("key1"),
            redis.get("key2"),
            connection_manager.mongo_client.db.collection.find_one({})
        ]
        
        results = await asyncio.gather(*tasks)
        self.write({"results": results})

Caching Results

import functools
import time

def cache_result(ttl=300):
    def decorator(func):
        cache = {}
        
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            # Build cache key
            cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
            
            # Check cache
            if cache_key in cache:
                data, timestamp = cache[cache_key]
                if time.time() - timestamp < ttl:
                    return data
            
            # Compute result
            result = await func(*args, **kwargs)
            
            # Store in cache
            cache[cache_key] = (result, time.time())
            
            return result
        
        return wrapper
    return decorator

class CachedHandler(tornado.web.RequestHandler):
    @cache_result(ttl=600)  # Cache for 10 minutes
    async def get_expensive_data(self):
        # Expensive operation
        await asyncio.sleep(2)
        return {"data": "expensive result"}
    
    async def get(self):
        result = await self.get_expensive_data()
        self.write(result)

Monitoring and Logging

Logging Configuration

import logging
import sys
from tornado.log import enable_pretty_logging

def setup_logging():
    # Configure root logger
    logging.getLogger().setLevel(logging.INFO)
    
    # Enable Tornado pretty logging
    enable_pretty_logging()
    
    # File logger
    file_handler = logging.FileHandler("app.log")
    file_formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    file_handler.setFormatter(file_formatter)
    
    # Add handler to root logger
    logging.getLogger().addHandler(file_handler)
    
    # Access logger configuration
    access_log = logging.getLogger("tornado.access")
    access_log.setLevel(logging.INFO)

class LoggingHandler(tornado.web.RequestHandler):
    def prepare(self):
        # Log incoming request
        logging.info(f"Request: {self.request.method} {self.request.uri}")
    
    def on_finish(self):
        # Log completed request
        logging.info(f"Response: {self.get_status()} in {self.request.request_time():.2f}s")

Performance Metrics

import time
import psutil
import tornado.web

class MetricsHandler(tornado.web.RequestHandler):
    def get(self):
        # System metrics
        cpu_percent = psutil.cpu_percent()
        memory = psutil.virtual_memory()
        
        # Application metrics
        active_connections = len(getattr(ChatWebSocket, 'connections', {}))
        
        metrics = {
            "timestamp": time.time(),
            "system": {
                "cpu_percent": cpu_percent,
                "memory_percent": memory.percent,
                "memory_available": memory.available,
            },
            "application": {
                "active_websockets": active_connections,
                "uptime": time.time() - self.application.start_time,
            }
        }
        
        self.write(metrics)

Frequently Asked Questions

What Is Tornado and What Is It Used For?

Tornado is an asynchronous web framework and HTTP server for Python, built to handle a large number of simultaneous connections. It excels at real‑time web applications, chats, API services, and high‑load systems.

How Does Tornado Differ from Flask and Django?

Tornado focuses on asynchrony and high performance. Unlike Flask and Django, which use a synchronous request‑handling model, Tornado employs non‑blocking I/O and can serve thousands of connections on a single thread.

Does Tornado Support Modern Python asyncio?

Yes. Starting with version 5.0, Tornado is fully compatible with asyncio, allowing you to use standard async/await constructs and integrate Tornado with other async libraries.

Can I Use Tornado to Build a REST API?

Absolutely. Tornado is well‑suited for REST APIs thanks to its flexibility and performance. It provides all the tools needed for handling HTTP verbs, JSON payloads, and authentication.

How Do I Secure a Tornado Application?

Tornado includes built‑in security mechanisms: CSRF protection, secure cookies, and HTTPS support. Additionally, you should validate input data, use parameterized database queries, and configure CORS properly.

Is Tornado Ready for Production?

Yes. Many large companies run Tornado in production. For a production deployment, it’s recommended to place Tornado behind a reverse proxy (e.g., Nginx), configure SSL, enable logging, and set up monitoring.

How Can I Scale a Tornado Application?

Tornado can be scaled vertically (by adding more CPU/RAM) and horizontally (by adding more server instances). For horizontal scaling, use a load balancer and external data stores such as Redis or MongoDB.

Are There Ready‑Made Solutions for Authentication?

Tornado provides basic authentication mechanisms and OAuth support. For more complex scenarios you can use third‑party libraries or implement a custom authentication system.

Tornado remains a powerful and up‑to‑date tool for building high‑performance web applications in Python. Its asynchronous nature and flexibility make it an excellent choice for modern web services, especially those that require handling many concurrent connections or real‑time interaction.

$