Python Tornado异步Web框架：核心概念与实践
Tornado是一个采用Python语言编写的Web框架和异步网络库，以其非阻塞式I/O和高性能而闻名。它最初由FriendFeed开发，旨在处理高并发、长连接的实时Web服务，能够轻松应对每秒数千个并发连接。
Tornado核心特点
与传统的同步阻塞型Web框架不同，Tornado采用非阻塞的工作模式和事件驱动架构，这使得它在处理大量并发连接时表现出色，尤其适合构建实时Web应用、API服务以及WebSockets服务。Tornado内部高效利用了操作系统提供的事件通知机制（如epoll），从而能够以极低的资源消耗管理大量活跃连接。
环境配置
安装Tornado非常简单，推荐使用pip包管理器：
pip install tornado
如果您使用Python 3，并且环境中同时存在Python 2，可能需要明确指定：
pip3 install tornado
初识Tornado：快速入门
下面是一个简单的Tornado应用，它会在访问特定路径时返回"Welcome to Tornado!"消息：
import tornado.web
import tornado.ioloop
import os
class WelcomeHandler(tornado.web.RequestHandler):
"""
处理 /welcome 路径的请求，返回欢迎信息。
"""
def get(self):
self.write("Welcome to Tornado!")
def make_simple_app():
"""
创建并返回一个简单的Tornado Web应用。
"""
return tornado.web.Application([
(r"/welcome", WelcomeHandler), # 定义路由，将 /welcome 映射到 WelcomeHandler
])
if __name__ == "__main__":
app = make_simple_app()
port = 9000
app.listen(port) # 监听指定端口
print(f"Tornado server listening on http://localhost:{port}/welcome")
# 启动I/O事件循环，使应用开始处理请求
tornado.ioloop.IOLoop.current().start()
运行上述脚本后，在浏览器中访问 http://localhost:9000/welcome，你将看到页面显示"Welcome to Tornado!"。
工作流程概述：
- 脚本启动，Tornado应用监听在指定端口。
- 客户端（如浏览器）发送HTTP请求到
/welcome。 - Tornado接收请求，根据路由规则将其分派给
WelcomeHandler类。 WelcomeHandler中的get方法被调用（因为是GET请求）。self.write("Welcome to Tornado!")将字符串内容写入HTTP响应体。- Tornado将响应发送回客户端。
异步非阻塞实践
Tornado的核心优势在于其异步非阻塞处理能力。使用async/await语法可以方便地编写异步代码，例如在处理请求时进行非阻塞的HTTP客户端调用。
import tornado.web
import tornado.ioloop
import tornado.httpclient
import asyncio
import json # 用于解析HTTP响应体
class AsyncFetchDataHandler(tornado.web.RequestHandler):
"""
演示如何使用Tornado的异步HTTP客户端发起请求。
"""
async def get(self):
start_message = "正在异步发起HTTP请求...<br>"
self.write(start_message)
http_client = tornado.httpclient.AsyncHTTPClient()
try:
# 异步调用外部API，不会阻塞当前IOLoop
# 示例使用 httpbin.org/get 返回请求的详细信息
response = await http_client.fetch("http://httpbin.org/get?param=tornado_test", request_timeout=10)
# 处理响应
status_code = response.code
response_body = response.body.decode('utf-8')
# 尝试解析JSON响应
try:
data = json.loads(response_body)
origin_ip = data.get('origin', '未知')
args_param = data.get('args', {}).get('param', '无')
result_message = (f"异步请求完成！<br>"
f"状态码: {status_code}<br>"
f"您的IP: {origin_ip}<br>"
f"请求参数 param: {args_param}<br>"
f"部分响应体: {response_body[:200]}...")
except json.JSONDecodeError:
result_message = (f"异步请求完成！<br>"
f"状态码: {status_code}<br>"
f"响应体非JSON格式，部分内容: {response_body[:200]}...")
self.write(result_message)
except tornado.httpclient.HTTPError as e:
self.write(f"HTTP请求错误: {e}")
except Exception as e:
self.write(f"发生未知错误: {e}")
def create_async_app():
"""
创建并返回一个演示异步功能的Tornado Web应用。
"""
return tornado.web.Application([
(r"/fetch_async", AsyncFetchDataHandler),
])
if __name__ == "__main__":
app_async = create_async_app()
async_port = 8001
app_async.listen(async_port)
print(f"异步示例服务器已启动，请访问 http://localhost:{async_port}/fetch_async")
tornado.ioloop.IOLoop.current().start()
在上述代码中，await http_client.fetch(...) 会暂停当前协程的执行，但不会阻塞整个Tornado事件循环。当HTTP响应返回时，协程会从暂停处继续执行，从而实现高效的并发处理。
路由系统详解
Tornado的路由系统负责将传入的URL路径映射到相应的请求处理器（RequestHandler）类。与一些将URL映射到函数的框架不同，Tornado的每个URL模式都对应一个类。
import tornado.web
import tornado.ioloop
class HomePageHandler(tornado.web.RequestHandler):
""" 首页处理器 """
def get(self):
self.write("欢迎访问Tornado主页！")
class ArticleHandler(tornado.web.RequestHandler):
""" 文章详情页处理器，支持路径参数 """
def get(self, article_id): # 通过位置参数获取URL中的匹配值
self.write(f"您正在查看文章，ID为：{article_id}")
class UserProfileHandler(tornado.web.RequestHandler):
""" 用户个人资料页处理器，使用命名捕获组 """
def get(self, username): # 通过命名参数获取URL中的匹配值
self.write(f"欢迎来到用户 {username} 的个人资料页。")
class AdminDashboardHandler(tornado.web.RequestHandler):
""" 管理员后台处理器（演示基于Host的路由） """
def get(self):
self.write("管理员后台仪表盘。")
def setup_routing_app():
"""
配置并返回带有多种路由规则的Tornado应用。
"""
# 主应用路由列表
main_routes = [
(r"/", HomePageHandler),
(r"/article/([0-9]+)", ArticleHandler), # 捕获一个或多个数字作为 article_id
(r"/user/(?P<username>[a-zA-Z0-9_]+)", UserProfileHandler), # 命名捕获组，捕获字母、数字、下划线
]
app = tornado.web.Application(main_routes)
# 演示基于Host的路由：针对特定域名处理请求
# 例如，如果请求的Host是 admin.example.com，则会应用此处的路由
# 注意：在本地测试需要修改 hosts 文件或使用代理
app.add_handlers(r"^(admin|dev)\.example\.com$", [ # 匹配 admin.example.com 或 dev.example.com
(r"/dashboard", AdminDashboardHandler), # 针对 /dashboard 路径
])
return app
if __name__ == "__main__":
router_app = setup_routing_app()
router_port = 8080
router_app.listen(router_port)
print(f"路由示例服务器已启动，监听端口 {router_port}")
print(f"访问示例:")
print(f" 主页: http://localhost:{router_port}/")
print(f" 文章: http://localhost:{router_port}/article/101")
print(f" 用户: http://localhost:{router_port}/user/alice_smith")
# 对于 add_handlers，您需要在 hosts 文件中添加映射或使用代理来测试 admin.example.com
tornado.ioloop.IOLoop.current().start()
路由规则使用正则表达式定义，括号 () 用于捕获URL中的片段，这些片段将作为参数传递给请求处理器的HTTP方法（如get、post）。命名捕获组 (?P 使得参数名称更明确。
模板引擎
Tornado内置了一个轻量级的模板引擎，语法类似于Django的模板。它支持模板继承、控制语句和表达语句，可以将动态数据渲染到HTML中。
- 控制语句：使用
{% ... %}包裹，例如{% if condition %},{% for item in list %}。这些语句需要通过{% end %}来结束对应的逻辑块。 - 表达语句：使用
{{ ... }}包裹，用于输出变量或表达式的值。 - 模板继承：通过
{% extends "base.html" %}和{% block block_name %} ... {% end %}实现代码复用。
为了演示，我们创建两个模板文件：base_template.html (基模板) 和 homepage.html (子模板)。
templates/base_template.html:
----------------------------
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>{% block page_title %}Tornado应用{% end %}</title>
<link href="{{ static_url("css/main.css") }}" rel="stylesheet">
{% block head_styles %}{% end %}
</head>
<body>
<header>
<h1>{% block header_content %}默认应用头部{% end %}</h1>
</header>
<nav>
<a href="/">主页</a> | <a href="/about">关于我们</a>
</nav>
<main>
{% block main_content %}{% end %}
</main>
<footer>
<p>© 2023 Tornado 示例</p>
</footer>
<script src="{{ static_url("js/common.js") }}"></script>
{% block body_scripts %}{% end %}
</body>
</html>
templates/homepage.html:
---------------------
{% extends 'base_template.html' %}
{% block page_title %}Tornado主页 - 动态内容{% end %}
{% block head_styles %}
<link href="{{ static_url("css/home.css") }}" rel="stylesheet">
{% end %}
{% block header_content %}欢迎访问我们的Tornado网站！{% end %}
{% block main_content %}
<h2>今日动态</h2>
{% if news_items %}
<ul>
{% for item in news_items %}
<li>{{ item }}</li>
{% end %}
</ul>
{% else %}
<p>暂无最新动态。</p>
{% end %}
{% end %}
{% block body_scripts %}
<script>
console.log("主页专属脚本已加载。");
</script>
{% end %}
import tornado.web
import tornado.ioloop
import os
class LandingPageHandler(tornado.web.RequestHandler):
"""
渲染主页模板，并传递动态数据。
"""
def get(self):
# 准备一些动态数据传递给模板
current_news = ["Tornado 6.0 发布", "异步编程新进展", "模板使用技巧"]
self.render('homepage.html', news_items=current_news)
def setup_template_app():
"""
设置并返回一个使用模板引擎的Tornado应用。
"""
# 定义模板和静态文件路径，相对于当前脚本
current_dir = os.path.dirname(__file__)
template_dir = os.path.join(current_dir, "templates")
static_dir = os.path.join(current_dir, "assets")
# 确保目录存在并创建一些演示文件
os.makedirs(template_dir, exist_ok=True)
os.makedirs(static_dir, exist_ok=True)
os.makedirs(os.path.join(static_dir, 'css'), exist_ok=True)
os.makedirs(os.path.join(static_dir, 'js'), exist_ok=True)
with open(os.path.join(template_dir, 'base_template.html'), 'w', encoding='utf-8') as f:
f.write("""<!DOCTYPE html><html lang="zh-CN"><head><meta charset="UTF-8"><title>{% block page_title %}Tornado应用{% end %}</title><link href="{{ static_url("css/main.css") }}" rel="stylesheet">{% block head_styles %}{% end %}</head><body><header><h1>{% block header_content %}默认应用头部{% end %}</h1></header><nav><a href="/">主页</a> | <a href="/about">关于我们</a></nav><main>{% block main_content %}{% end %}</main><footer><p>© 2023 Tornado 示例</p></footer><script src="{{ static_url("js/common.js") }}"></script>{% block body_scripts %}{% end %}</body></html>""")
with open(os.path.join(template_dir, 'homepage.html'), 'w', encoding='utf-8') as f:
f.write("""{% extends 'base_template.html' %}\n{% block page_title %}Tornado主页 - 动态内容{% end %}\n{% block head_styles %}\n <link href="{{ static_url("css/home.css") }}" rel="stylesheet">\n{% end %}\n{% block header_content %}欢迎访问我们的Tornado网站！{% end %}\n{% block main_content %}\n <h2>今日动态</h2>\n {% if news_items %}\n <ul>\n {% for item in news_items %}\n <li>{{ item }}</li>\n {% end %}\n </ul>\n {% else %}\n <p>暂无最新动态。</p>\n {% end %}\n{% end %}\n{% block body_scripts %}\n <script>\n console.log("主页专属脚本已加载。");\n </script>\n{% end %}""")
with open(os.path.join(static_dir, 'css', 'main.css'), 'w') as f: f.write('body { font-family: Arial, sans-serif; margin: 0; padding: 0; background-color: #f4f4f4; } header, footer { background-color: #333; color: white; padding: 10px 20px; text-align: center; } nav { background-color: #eee; padding: 10px; text-align: center; } main { padding: 20px; }')
with open(os.path.join(static_dir, 'css', 'home.css'), 'w') as f: f.write('h2 { color: #007bff; } ul { list-style-type: disc; margin-left: 20px; }')
with open(os.path.join(static_dir, 'js', 'common.js'), 'w') as f: f.write('console.log("Common JS loaded.");')
settings = {
'template_path': template_dir,
'static_path': static_dir,
'static_url_prefix': '/assets/', # 静态文件URL前缀
}
app = tornado.web.Application([
(r"/", LandingPageHandler),
], **settings)
return app
if __name__ == "__main__":
template_app = setup_template_app()
template_port = 8081
template_app.listen(template_port)
print(f"模板示例服务器已启动，请访问 http://localhost:{template_port}/")
tornado.ioloop.IOLoop.current().start()
模板内置功能与自定义UI组件
Tornado模板引擎提供了一系列内置的函数和变量，方便在模板中使用：
escape: HTML转义，防止XSS攻击。static_url(path): 生成静态文件的URL，会自动处理版本号以实现缓存。xsrf_form_html(): 生成XSRF保护所需的隐藏表单字段。handler,request,current_user,locale: 访问当前的RequestHandler对象及其相关属性。datetime: Python的datetime模块。
除了内置功能，Tornado还支持通过UIMethods和UIModules自定义模板逻辑和组件，类似于Django的自定义标签或组件：
- UIMethods：普通的Python函数，可以在模板中作为函数调用。
- UIModules：继承自
tornado.web.UIModule的类，通常用于渲染更复杂的、可重用的UI片段。
import tornado.web
import tornado.ioloop
import tornado.escape
import os
# --- 自定义UIMethods ---
def page_footer_info(handler):
"""
一个简单的UIMethod，用于在页面底部显示信息。
handler 参数是当前的 RequestHandler 实例。
"""
return "<span>Powered by Tornado Framework.</span>"
# --- 自定义UIModules ---
class UserBadge(tornado.web.UIModule):
"""
一个简单的UIModule，用于显示用户徽章。
"""
def render(self, username, level):
# render 方法返回要在模板中插入的HTML字符串
return tornado.escape.xhtml_escape(f"""
<div class="user-badge">
<strong>{username}</strong> (等级: {level})
</div>
""")
class CustomUIHandler(tornado.web.RequestHandler):
"""
渲染包含自定义UIMethod和UIModule的模板。
"""
def get(self):
self.render('ui_showcase.html', visitor_name="普通用户", visitor_level=3)
def setup_custom_ui_app():
"""
设置并返回一个演示自定义UIMethod和UIModule的Tornado应用。
"""
current_dir = os.path.dirname(__file__)
template_dir = os.path.join(current_dir, "templates_ui")
static_dir = os.path.join(current_dir, "assets_ui")
os.makedirs(template_dir, exist_ok=True)
os.makedirs(static_dir, exist_ok=True)
os.makedirs(os.path.join(static_dir, 'css'), exist_ok=True)
with open(os.path.join(template_dir, 'ui_showcase.html'), 'w', encoding='utf-8') as f:
f.write("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>自定义UI组件示例</title>
<link href="{{ static_url("css/ui_styles.css") }}" rel="stylesheet">
</head>
<body>
<h1>Tornado 自定义 UI 示例</h1>
<h2>UIMethod 调用</h2>
<div class="uimethod-output">
{{ page_footer_info() }}
</div>
<h2>UIModule 调用</h2>
<div class="uimodule-output">
{% module UserBadge(visitor_name, visitor_level) %}
{% module UserBadge("高级会员", 8) %}
</div>
</body>
</html>
""")
with open(os.path.join(static_dir, 'css', 'ui_styles.css'), 'w') as f:
f.write(".user-badge { border: 1px solid #007bff; background-color: #e6f2ff; padding: 8px 12px; margin: 5px; display: inline-block; border-radius: 5px; font-size: 0.9em; } h2 { color: #333; } .uimethod-output, .uimodule-output { margin-bottom: 20px; padding: 10px; border: 1px dashed #ccc; }")
settings = {
'template_path': template_dir,
'static_path': static_dir,
'static_url_prefix': '/assets_ui/',
'ui_methods': {
'page_footer_info': page_footer_info, # 注册自定义UIMethod
},
'ui_modules': {
'UserBadge': UserBadge, # 注册自定义UIModule
},
}
app = tornado.web.Application([
(r"/ui_components", CustomUIHandler),
], **settings)
return app
if __name__ == "__main__":
ui_app = setup_custom_ui_app()
ui_port = 8082
ui_app.listen(ui_port)
print(f"自定义UI示例服务器已启动，请访问 http://localhost:{ui_port}/ui_components")
tornado.ioloop.IOLoop.current().start()
实用功能
1. 静态文件服务
Tornado可以方便地配置静态文件的服务。通过设置static_path指定静态文件目录，static_url_prefix指定URL前缀，Tornado会自动处理文件的访问请求。
Tornado还会为静态文件自动生成带内容哈希的版本化URL（例如 /static/css/style.css?v=abcdef123456），这有助于浏览器缓存管理和CDN部署。
import tornado.web
import tornado.ioloop
import os
class StaticFileDemoHandler(tornado.web.RequestHandler):
"""
渲染一个演示静态文件加载的页面。
"""
def get(self):
self.render("static_display.html")
def setup_static_app():
"""
设置并返回一个配置了静态文件服务的Tornado应用。
"""
current_dir = os.path.dirname(__file__)
template_dir = os.path.join(current_dir, "templates_static")
static_content_dir = os.path.join(current_dir, "web_assets") # 静态文件目录名自定义为 web_assets
os.makedirs(template_dir, exist_ok=True)
os.makedirs(static_content_dir, exist_ok=True)
os.makedirs(os.path.join(static_content_dir, 'css'), exist_ok=True)
os.makedirs(os.path.join(static_content_dir, 'js'), exist_ok=True)
os.makedirs(os.path.join(static_content_dir, 'images'), exist_ok=True)
with open(os.path.join(template_dir, 'static_display.html'), 'w', encoding='utf-8') as f:
f.write("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>静态文件加载演示</title>
<link href="{{ static_url("css/theme.css") }}" rel="stylesheet">
</head>
<body>
<h1>Tornado 静态文件加载测试</h1>
<img src="{{ static_url("images/sample_icon.png") }}" alt="示例图标" width="80">
<p>这段文字的样式由外部CSS文件控制。</p>
<script src="{{ static_url("js/app_script.js") }}"></script>
</body>
</html>
""")
with open(os.path.join(static_content_dir, 'css', 'theme.css'), 'w') as f:
f.write("body { font-family: 'Open Sans', sans-serif; background-color: #e9ecef; color: #495057; margin: 25px; } h1 { color: #007bff; border-bottom: 2px solid #007bff; padding-bottom: 10px; } img { border: 1px solid #adb5bd; border-radius: 5px; }")
with open(os.path.join(static_content_dir, 'js', 'app_script.js'), 'w') as f:
f.write("console.log('App specific JavaScript loaded!'); alert('Hello from Tornado static JS!');")
# 创建一个空的图片文件作为占位符
with open(os.path.join(static_content_dir, 'images', 'sample_icon.png'), 'w') as f:
f.write('')
settings = {
'template_path': template_dir,
'static_path': static_content_dir,
'static_url_prefix': '/static_files/', # 自定义静态文件URL前缀
'debug': True, # 启用 debug 模式，有助于开发时静态文件修改的实时反映
}
app = tornado.web.Application([
(r"/show_static", StaticFileDemoHandler),
], **settings)
return app
if __name__ == "__main__":
static_app = setup_static_app()
static_port = 8083
static_app.listen(static_port)
print(f"静态文件示例服务器已启动，请访问 http://localhost:{static_port}/show_static")
tornado.ioloop.IOLoop.current().start()
2. 跨站请求伪造 (CSRF) 保护
CSRF是一种常见的Web安全漏洞，攻击者诱导用户点击恶意链接，利用用户已登录的身份执行非用户本意的操作。Tornado通过xsrf_cookies配置提供了内置的CSRF保护机制。
当xsrf_cookies设置为True时，Tornado会：
- 在用户的浏览器中设置一个名为
_xsrf的Cookie。 - 在渲染表单时，通过
{% raw xsrf_form_html() %}（需要使用raw以避免模板引擎再次转义）在HTML中注入一个隐藏的_xsrf字段。 - 在处理POST、PUT、DELETE等请求时，验证请求中
_xsrf字段的值是否与Cookie中的值匹配。
import tornado.web
import tornado.ioloop
import os
import json # 用于AJAX示例解析请求体
import uuid # 用于生成 cookie_secret
class StandardFormHandler(tornado.web.RequestHandler):
"""
处理标准HTML表单的CSRF保护演示。
"""
def get(self):
self.render("simple_form_csrf.html")
def post(self):
# Tornado会自动验证_xsrf token
action = self.get_argument("action_type", "未知操作")
self.write(f"表单提交成功！执行操作：'{action}'。CSRF令牌已验证。")
class AjaxFormHandler(tornado.web.RequestHandler):
"""
处理AJAX请求的CSRF保护演示。
"""
def get(self):
self.render("ajax_form_csrf.html")
def post(self):
# 对于AJAX请求，_xsrf令牌通常通过请求头或请求体发送
# Tornado会进行自动验证
try:
request_data = json.loads(self.request.body)
message = request_data.get("message", "无消息")
self.write(f"AJAX请求处理成功！消息：'{message}'。CSRF令牌已验证。")
self.set_header("Content-Type", "application/json")
except json.JSONDecodeError:
raise tornado.web.HTTPError(400, reason="Invalid JSON body")
def setup_csrf_app():
"""
设置并返回一个启用CSRF保护的Tornado应用。
"""
current_dir = os.path.dirname(__file__)
template_dir = os.path.join(current_dir, "templates_csrf")
os.makedirs(template_dir, exist_ok=True)
with open(os.path.join(template_dir, 'simple_form_csrf.html'), 'w', encoding='utf-8') as f:
f.write("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>标准表单 (CSRF保护)</title>
</head>
<body>
<h1>标准表单提交演示</h1>
<form action="/submit_action" method="post">
{% raw xsrf_form_html() %} {# 插入CSRF隐藏字段 #}
<p>操作类型: <input type="text" name="action_type" value="更新配置"></p>
<p><input type="submit" value="提交表单"></p>
</form>
</body>
</html>
""")
with open(os.path.join(template_dir, 'ajax_form_csrf.html'), 'w', encoding='utf-8') as f:
f.write("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>AJAX提交 (CSRF保护)</title>
</head>
<body>
<h1>AJAX 数据提交演示</h1>
<input type="text" id="ajaxMessageInput" placeholder="输入您的消息">
<button onclick="sendAjaxData()">发送消息</button>
<div id="ajaxResponse" style="margin-top: 10px; color: blue;"></div>
<script>
function getCsrfToken() {
const nameEQ = "_xsrf=";
const ca = document.cookie.split(';');
for(let i=0; i < ca.length; i++) {
let c = ca[i];
while (c.charAt(0) === ' ') c = c.substring(1, c.length);
if (c.indexOf(nameEQ) === 0) return c.substring(nameEQ.length, c.length);
}
return null;
}
async function sendAjaxData() {
const message = document.getElementById('ajaxMessageInput').value;
const xsrfToken = getCsrfToken(); // 从Cookie获取_xsrf令牌
if (!xsrfToken) {
alert("未找到CSRF令牌，请确保 xsrf_cookies 已启用。");
return;
}
const responseDiv = document.getElementById('ajaxResponse');
responseDiv.textContent = "正在发送...";
try {
const response = await fetch('/submit_ajax_data', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'X-XSRFToken': xsrfToken // 将令牌放在自定义请求头中
},
body: JSON.stringify({ message: message })
});
const result = await response.text();
responseDiv.textContent = '服务器响应: ' + result;
} catch (error) {
console.error('AJAX请求错误:', error);
responseDiv.textContent = 'AJAX请求失败。';
}
}
</script>
</body>
</html>
""")
settings = {
'template_path': template_dir,
'xsrf_cookies': True, # 启用XSRF保护
'cookie_secret': str(uuid.uuid4()).replace('-', ''), # 必须提供一个安全的 cookie_secret
}
app = tornado.web.Application([
(r"/form_page", StandardFormHandler),
(r"/submit_action", StandardFormHandler),
(r"/ajax_page", AjaxFormHandler),
(r"/submit_ajax_data", AjaxFormHandler),
], **settings)
return app
if __name__ == "__main__":
csrf_app = setup_csrf_app()
csrf_port = 8084
csrf_app.listen(csrf_port)
print(f"CSRF示例服务器已启动，请访问:")
print(f" 标准表单: http://localhost:{csrf_port}/form_page")
print(f" AJAX表单: http://localhost:{csrf_port}/ajax_page")
tornado.ioloop.IOLoop.current().start()
3. Cookie管理
Tornado提供了方便的方法来设置和读取HTTP Cookie。此外，为了防止Cookie被伪造或篡改，Tornado还支持签名Cookie。
- 普通Cookie：使用
set_cookie(name, value, ...)和get_cookie(name, default=None)。 - 签名Cookie：使用
set_secure_cookie(name, value, ...)和get_secure_cookie(name, default=None)。使用签名Cookie时，应用必须配置一个cookie_secret密钥。Tornado会用这个密钥对Cookie值进行签名，并在读取时验证签名，确保其未被篡改。
import tornado.web
import tornado.ioloop
import os
import base64
import uuid # 用于生成 cookie_secret
class StandardCookieHandler(tornado.web.RequestHandler):
""" 演示普通Cookie的设置与读取 """
def get(self):
# 读取一个名为 'visit_count' 的Cookie
current_visits = int(self.get_cookie("visit_count", "0"))
new_visits = current_visits + 1
# 设置Cookie，有效期为1天
self.set_cookie("visit_count", str(new_visits), expires_days=1)
self.write(f"您已访问此页面 {new_visits} 次 (普通Cookie)。")
class SignedCookieHandler(tornado.web.RequestHandler):
""" 演示签名Cookie的设置与读取 """
def get(self):
# 尝试读取一个名为 'user_session_token' 的签名Cookie
user_token = self.get_secure_cookie("user_session_token")
if user_token:
self.write(f"检测到安全Cookie：令牌 {user_token.decode()}。")
else:
# 如果没有，则设置一个新的签名Cookie
new_token = str(uuid.uuid4())
self.set_secure_cookie("user_session_token", new_token, expires_days=7)
self.write(f"已设置新的安全Cookie：令牌 {new_token}。")
class UserAuthBaseHandler(tornado.web.RequestHandler):
"""
基础处理器，提供 get_current_user 方法，用于认证装饰器。
"""
def get_current_user(self):
# 从签名Cookie中获取用户标识
return self.get_secure_cookie("authenticated_user")
class UserDashboardHandler(UserAuthBaseHandler):
"""
一个需要认证才能访问的页面。
"""
@tornado.web.authenticated # 使用 @authenticated 装饰器进行认证
def get(self):
# 如果用户已认证，current_user 将是 get_current_user 的返回值
current_user_id = self.current_user.decode()
self.write(f"欢迎来到用户仪表盘，用户ID: {current_user_id}。")
self.write("<p><a href='/logout_user'>退出登录</a></p>")
class UserLoginHandler(tornado.web.RequestHandler):
"""
用户登录页面。
"""
def get(self):
self.render("login_with_secure_cookie.html", message="")
def post(self):
username = self.get_argument("username", "")
password = self.get_argument("password", "")
# 简单的模拟登录验证
if username == "admin" and password == "123456":
# 登录成功，设置签名Cookie
self.set_secure_cookie("authenticated_user", username, expires_days=1)
self.redirect("/dashboard")
else:
self.render("login_with_secure_cookie.html", message="用户名或密码错误。")
class UserLogoutHandler(tornado.web.RequestHandler):
"""
用户退出登录。
"""
def get(self):
self.clear_cookie("authenticated_user") # 清除认证Cookie
self.redirect("/login_page")
def setup_cookie_app():
"""
设置并返回一个包含Cookie管理和认证功能的Tornado应用。
"""
current_dir = os.path.dirname(__file__)
template_dir = os.path.join(current_dir, "templates_cookie")
os.makedirs(template_dir, exist_ok=True)
with open(os.path.join(template_dir, 'login_with_secure_cookie.html'), 'w', encoding='utf-8') as f:
f.write("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>用户登录</title>
</head>
<body>
<h1>系统登录</h1>
<p style="color:red;">{{ message }}</p>
<form action="/login_page" method="post">
{% raw xsrf_form_html() %} {# 启用XSRF保护 #}
<p>用户名: <input type="text" name="username" value="admin"></p>
<p>密码: <input type="password" name="password" value="123456"></p>
<p><input type="submit" value="登录"></p>
</form>
</body>
</html>
""")
# 必须提供一个强加密密钥用于Cookie签名
# 生产环境中应使用更安全的密钥管理方式
cookie_secret = base64.b64encode(uuid.uuid4().bytes + uuid.uuid4().bytes).decode('utf-8')
settings = {
'cookie_secret': cookie_secret,
'xsrf_cookies': True, # 推荐在所有表单中启用XSRF保护
'login_url': '/login_page', # 当 @authenticated 装饰器失败时重定向的URL
'template_path': template_dir,
}
app = tornado.web.Application([
(r"/simple_cookie", StandardCookieHandler),
(r"/secure_session_cookie", SignedCookieHandler),
(r"/login_page", UserLoginHandler),
(r"/dashboard", UserDashboardHandler),
(r"/logout_user", UserLogoutHandler),
], **settings)
return app
if __name__ == "__main__":
cookie_app = setup_cookie_app()
cookie_port = 8085
cookie_app.listen(cookie_port)
print(f"Cookie示例服务器已启动，请访问:")
print(f" 普通Cookie: http://localhost:{cookie_port}/simple_cookie")
print(f" 签名Cookie: http://localhost:{cookie_port}/secure_session_cookie")
print(f" 登录页面: http://localhost:{cookie_port}/login_page (登录后访问 /dashboard)")
tornado.ioloop.IOLoop.current().start()
4. AJAX文件上传
处理文件上传是Web应用常见需求。Tornado可以轻松处理通过AJAX发送的multipart/form-data类型的文件上传请求。
import tornado.web
import tornado.ioloop
import os
import json
class AjaxFileUploadHandler(tornado.web.RequestHandler):
"""
处理AJAX文件上传请求。
"""
def get(self):
self.render("file_upload_form.html")
def post(self):
# 'uploaded_files_field' 是HTML表单中input元素的 name 属性值
uploaded_files = self.request.files.get("uploaded_files_field", [])
if not uploaded_files:
# 如果没有文件上传，返回错误信息
self.set_status(400)
self.write(json.dumps({"status": "error", "message": "未检测到上传文件。"}))
self.set_header("Content-Type", "application/json")
return
response_messages = []
# 定义文件保存路径
upload_dir = os.path.join(os.path.dirname(__file__), "uploaded_files")
os.makedirs(upload_dir, exist_ok=True) # 确保上传目录存在
for file_meta in uploaded_files:
filename = file_meta['filename']
filepath = os.path.join(upload_dir, filename)
try:
# 以二进制写入模式保存文件
with open(filepath, 'wb') as new_file:
new_file.write(file_meta['body'])
response_messages.append(f"文件 '{filename}' 上传成功并保存到 '{upload_dir}'。")
except Exception as e:
response_messages.append(f"文件 '{filename}' 上传失败: {e}")
self.write(json.dumps({"status": "success", "messages": response_messages}))
self.set_header("Content-Type", "application/json")
def setup_upload_app():
"""
设置并返回一个包含AJAX文件上传功能的Tornado应用。
"""
current_dir = os.path.dirname(__file__)
template_dir = os.path.join(current_dir, "templates_upload")
os.makedirs(template_dir, exist_ok=True)
with open(os.path.join(template_dir, 'file_upload_form.html'), 'w', encoding='utf-8') as f:
f.write("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>AJAX 文件上传</title>
</head>
<body>
<h1>使用 Fetch API 进行 AJAX 文件上传</h1>
<input type="file" id="fileSelector" name="uploadInput" multiple>
<button onclick="performFileUpload()">开始上传</button>
<div id="uploadStatusArea" style="margin-top: 15px; border: 1px solid #ccc; padding: 10px; min-height: 50px;">
上传状态信息将显示在这里。
</div>
<script>
async function performFileUpload() {
const fileInput = document.getElementById('fileSelector');
const filesToUpload = fileInput.files;
if (filesToUpload.length === 0) {
alert("请选择至少一个文件进行上传！");
return;
}
const formData = new FormData();
// 将所有选中的文件添加到 FormData 对象中
// 'uploaded_files_field' 必须与后端RequestHandler中 request.files.get() 的键名一致
for (let i = 0; i < filesToUpload.length; i++) {
formData.append('uploaded_files_field', filesToUpload[i]);
}
const statusArea = document.getElementById('uploadStatusArea');
statusArea.innerHTML = '<p style="color: gray;">文件上传中，请稍候...</p>';
try {
const response = await fetch('/upload_endpoint_path', {
method: 'POST',
body: formData // 使用 FormData 时，无需手动设置 Content-Type 头，浏览器会自动设置
});
const result = await response.json();
if (result.status === 'success') {
statusArea.innerHTML = '<h3 style="color: green;">上传成功！</h3><ul>' +
result.messages.map(msg => '<li>' + msg + '</li>').join('') +
'</ul>';
} else {
statusArea.innerHTML = '<h3 style="color: red;">上传失败！</h3><p>' + result.message + '</p>';
}
} catch (error) {
console.error('文件上传请求失败:', error);
statusArea.innerHTML = '<h3 style="color: red;">请求过程中发生错误。</h3>';
}
}
</script>
</body>
</html>
""")
app_settings = {
'template_path': template_dir,
'debug': True, # 开发模式下便于调试
}
app = tornado.web.Application([
(r"/upload_page_entry", AjaxFileUploadHandler),
(r"/upload_endpoint_path", AjaxFileUploadHandler),
], **app_settings)
return app
if __name__ == "__main__":
upload_app = setup_upload_app()
upload_port = 8086
upload_app.listen(upload_port)
print(f"文件上传示例服务器已启动，请访问 http://localhost:{upload_port}/upload_page_entry")
tornado.ioloop.IOLoop.current().start()
扩展功能
1. 自定义会话 (Session)
Tornado本身不提供内置的会话管理机制，但我们可以利用Python的特殊方法（如__getitem__, __setitem__, __delitem__）来实现自定义的会话类。
下面是一个基于内存的简单会话实现。在生产环境中，内存会话不适用于多进程或分布式部署，通常会将其存储在Redis、Memcached等外部存储中。
import tornado.web
import tornado.ioloop
import os
import uuid
import time
import hashlib
# --- 1. 内存会话实现 (仅用于演示，不适合生产环境) ---
# 全局字典，用于存储所有会话数据，键为session ID
global_session_storage = {}
def generate_unique_session_id():
""" 生成一个基于SHA1的唯一会话ID """
return hashlib.sha1(os.urandom(16) + str(time.time()).encode('utf-8')).hexdigest()
class MemorySession:
"""
一个简单的基于内存的会话类，通过 RequestHandler 关联。
"""
SESSION_COOKIE_KEY = "_app_session_id" # 存储会话ID的Cookie名称
def __init__(self, request_handler):
self.handler = request_handler
session_id = request_handler.get_cookie(self.SESSION_COOKIE_KEY)
if not session_id or session_id not in global_session_storage:
# 如果没有找到会话ID或会话已过期，则创建新会话
self._session_id = generate_unique_session_id()
global_session_storage[self._session_id] = {} # 初始化为空字典
# 设置会话Cookie，通常设置过期时间
request_handler.set_cookie(self.SESSION_COOKIE_KEY, self._session_id, expires_days=1)
else:
self._session_id = session_id
def __getitem__(self, key):
""" 获取会话中的值 """
return global_session_storage[self._session_id].get(key)
def __setitem__(self, key, value):
""" 设置会话中的值 """
global_session_storage[self._session_id][key] = value
def __delitem__(self, key):
""" 删除会话中的某个值 """
if key in global_session_storage[self._session_id]:
del global_session_storage[self._session_id][key]
def clear(self):
""" 清空当前会话所有数据 """
if self._session_id in global_session_storage:
del global_session_storage[self._session_id]
class SessionAwareBaseHandler(tornado.web.RequestHandler):
"""
所有需要会话功能的RequestHandler都应继承此基类。
"""
def prepare(self):
# 在处理请求前初始化会话对象
self.user_session = MemorySession(self)
def get_current_user(self):
# 供 @tornado.web.authenticated 装饰器使用
return self.user_session.get("logged_user_id")
class UserLoginPageHandler(SessionAwareBaseHandler):
""" 用户登录页面 """
def get(self):
self.render("session_login_form.html", status_message="")
def post(self):
username = self.get_argument("uname", "")
password = self.get_argument("pwd", "")
if username == "session_user" and password == "secure_pass":
# 登录成功，将会话信息存储到自定义会话中
self.user_session["logged_user_id"] = username
self.user_session["login_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
self.redirect("/dashboard_session")
else:
self.render("session_login_form.html", status_message="用户名或密码错误。")
class UserSessionDashboardHandler(SessionAwareBaseHandler):
""" 登录后才能访问的仪表盘页面 """
@tornado.web.authenticated
def get(self):
user_id = self.user_session.get("logged_user_id")
login_time = self.user_session.get("login_time", "N/A")
self.write(f"欢迎您，{user_id}！您上次登录时间是: {login_time}")
self.write("<p><a href='/logout_session'>退出登录</a></p>")
class UserSessionLogoutHandler(SessionAwareBaseHandler):
""" 退出登录并清除会话 """
def get(self):
self.user_session.clear() # 清除内存中的会话数据
self.clear_cookie(MemorySession.SESSION_COOKIE_KEY) # 清除浏览器中的会话ID Cookie
self.redirect("/login_session")
def setup_custom_session_app():
"""
设置并返回一个包含自定义会话功能的Tornado应用。
"""
current_dir = os.path.dirname(__file__)
template_dir = os.path.join(current_dir, "templates_session")
os.makedirs(template_dir, exist_ok=True)
with open(os.path.join(template_dir, 'session_login_form.html'), 'w', encoding='utf-8') as f:
f.write("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>自定义会话登录</title>
</head>
<body>
<h1>登录 (自定义会话)</h1>
<p style="color:red;">{{ status_message }}</p>
<form action="/login_session" method="post">
<p>用户名: <input type="text" name="uname" value="session_user"></p>
<p>密码: <input type="password" name="pwd" value="secure_pass"></p>
<p><input type="submit" value="登录"></p>
</form>
</body>
</html>
""")
app_settings = {
'template_path': template_dir,
'login_url': '/login_session', # @authenticated 装饰器失败时的重定向
# 即使自定义会话不直接使用，Tornado的其他功能（如签名Cookie）可能需要此密钥
'cookie_secret': hashlib.sha256(os.urandom(64)).hexdigest(),
}
app = tornado.web.Application([
(r"/login_session", UserLoginPageHandler),
(r"/dashboard_session", UserSessionDashboardHandler),
(r"/logout_session", UserSessionLogoutHandler),
], **app_settings)
return app
if __name__ == "__main__":
session_app = setup_custom_session_app()
session_port = 8087
session_app.listen(session_port)
print(f"自定义会话示例服务器已启动，请访问 http://localhost:{session_port}/login_session")
tornado.ioloop.IOLoop.current().start()
2. 表单验证框架
在Web开发中，对用户提交的数据进行验证是至关重要的一环。Tornado本身没有像Django或Flask-WTF那样内置的表单验证库，但我们可以构建一个灵活的自定义验证框架。
这个框架包含BaseField基类和派生出的具体字段类型（如StringField, EmailField, IntegerField），以及一个用于组合这些字段的CustomForm类。它能够实现：
- 定义字段的验证规则（如是否必填、最小/最大长度、正则表达式）。
- 自动从请求中获取数据并进行验证。
- 收集验证结果和错误信息。
- 将验证后的"干净"数据（
cleaned_data）提供给业务逻辑。
import tornado.web
import tornado.ioloop
import re
import os
# --- 自定义验证框架 ---
class BaseField:
"""
所有表单字段的基类。
"""
def __init__(self, required=True, error_messages=None):
self.required = required
self.value = None
self.errors = []
self._name = None # 字段的名称，在 validate 方法中设置
self.error_messages = error_messages or {} # 自定义错误消息
def validate(self, field_name, input_value):
"""
验证字段的方法。
:param field_name: 字段的名称 (对应HTML input的name属性)。
:param input_value: 从请求中获取的原始输入值。
:return: 布尔值，表示验证是否通过。
"""
self._name = field_name
self.errors = [] # 每次验证前清除旧错误
if not input_value and self.required:
self.errors.append(self.error_messages.get('required', f"{field_name} 是必填项。"))
return False
if input_value is not None: # 仅当有值时才运行具体验证逻辑
return self._run_validation_logic(input_value)
self.value = input_value # 可选字段如果无值，则保存 None 或空字符串
return True
def _run_validation_logic(self, value):
"""
具体的验证逻辑，由子类实现。
:param value: 待验证的值。
:return: 布尔值。
"""
self.value = value # 默认情况下直接接受值
return True
class StringField(BaseField):
""" 字符串字段验证 """
def __init__(self, required=True, min_length=None, max_length=None, error_messages=None):
super().__init__(required, error_messages)
self.min_length = min_length
self.max_length = max_length
def _run_validation_logic(self, value):
if not isinstance(value, str):
self.errors.append(self.error_messages.get('type', f"{self._name} 必须是字符串。"))
return False
if self.min_length is not None and len(value) < self.min_length:
self.errors.append(self.error_messages.get('min_length', f"{self._name} 至少需要 {self.min_length} 个字符。"))
if self.max_length is not None and len(value) > self.max_length:
self.errors.append(self.error_messages.get('max_length', f"{self._name} 最多允许 {self.max_length} 个字符。"))
if not self.errors:
self.value = value
return True
return False
class EmailField(StringField):
""" 邮箱字段验证 """
EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
def _run_validation_logic(self, value):
# 首先运行父类(StringField)的验证
if not super()._run_validation_logic(value):
return False
if not self.EMAIL_REGEX.match(value):
self.errors.append(self.error_messages.get('invalid', f"{self._name} 格式不正确。"))
return False
self.value = value
return True
class IntegerField(BaseField):
""" 整数字段验证 """
def __init__(self, required=True, min_value=None, max_value=None, error_messages=None):
super().__init__(required, error_messages)
self.min_value = min_value
self.max_value = max_value
def _run_validation_logic(self, value):
try:
int_value = int(value)
if self.min_value is not None and int_value < self.min_value:
self.errors.append(self.error_messages.get('min_value', f"{self._name} 必须大于等于 {self.min_value}。"))
if self.max_value is not None and int_value > self.max_value:
self.errors.append(self.error_messages.get('max_value', f"{self._name} 必须小于等于 {self.max_value}。"))
if not self.errors:
self.value = int_value
return True
except ValueError:
self.errors.append(self.error_messages.get('invalid', f"{self._name} 必须是有效的整数。"))
return False
class CustomForm:
"""
表单基类，用于组合多个字段并进行整体验证。
"""
def __init__(self):
self._fields = {}
self.is_valid = False
self.cleaned_data = {} # 验证通过后的"干净"数据
self.form_errors = {} # 包含所有字段的错误信息
# 自动发现子类中定义的 BaseField 实例
for name in dir(self):
if not name.startswith('_'): # 避免内部属性
attribute = getattr(self, name)
if isinstance(attribute, BaseField):
self._fields[name] = attribute
def validate_request(self, request_handler):
"""
从 RequestHandler 中提取数据并验证整个表单。
:param request_handler: 当前的 RequestHandler 实例。
:return: 布尔值，表示表单是否完全验证通过。
"""
self.is_valid = True
self.cleaned_data = {}
self.form_errors = {}
for field_name, field_obj in self._fields.items():
# Tornado 的 get_argument 默认返回字符串或 None (如果参数不存在)
input_value = request_handler.get_argument(field_name, None)
if field_obj.validate(field_name, input_value):
self.cleaned_data[field_name] = field_obj.value
else:
self.is_valid = False
self.form_errors[field_name] = field_obj.errors
return self.is_valid
# --- 应用特定的表单定义 ---
class UserRegistrationForm(CustomForm):
"""
用户注册表单，包含用户名、邮箱和年龄字段。
"""
def __init__(self):
super().__init__()
self.username = StringField(required=True, min_length=4, max_length=20,
error_messages={'required': '用户名是必填项', 'min_length': '用户名至少4个字符', 'max_length': '用户名最多20个字符'})
self.email = EmailField(required=True,
error_messages={'required': '邮箱是必填项', 'invalid': '邮箱格式不正确'})
self.age = IntegerField(required=False, min_value=18, max_value=100,
error_messages={'min_value': '年龄必须满18岁', 'max_value': '年龄不能超过100岁', 'invalid': '年龄必须是数字'})
class RegistrationHandler(tornado.web.RequestHandler):
"""
处理用户注册的请求处理器。
"""
def get(self):
# 首次访问，渲染空表单
self.render("registration_form_page.html", form_data={}, errors={})
def post(self):
form = UserRegistrationForm()
if form.validate_request(self):
# 表单验证通过
# 可以在这里处理 form.cleaned_data (如保存到数据库)
self.write(f"<h1>注册成功！</h1><p>提交的数据: {form.cleaned_data}</p>")
else:
# 表单验证失败，重新渲染表单，显示错误信息和用户原输入
# request.arguments 是一个字典，其值为列表，如 {'username': ['test']}
# 模板中需要处理这种格式
self.render("registration_form_page.html", form_data=self.request.arguments, errors=form.form_errors)
def setup_validation_app():
"""
设置并返回一个包含表单验证功能的Tornado应用。
"""
current_dir = os.path.dirname(__file__)
template_dir = os.path.join(current_dir, "templates_validation")
os.makedirs(template_dir, exist_ok=True)
with open(os.path.join(template_dir, 'registration_form_page.html'), 'w', encoding='utf-8') as f:
f.write("""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>用户注册表单</title>
<style>
.error { color: red; font-size: 0.85em; margin-left: 10px; }
.form-group { margin-bottom: 12px; }
label { display: inline-block; width: 85px; text-align: right; margin-right: 10px; }
input[type="text"], input[type="email"] { padding: 6px; border: 1px solid #ccc; border-radius: 4px; width: 220px; }
button { padding: 8px 15px; background-color: #007bff; color: white; border: none; border-radius: 4px; cursor: pointer; }
button:hover { background-color: #0056b3; }
</style>
</head>
<body>
<h1>用户注册</h1>
<form action="/user_register" method="post">
<div class="form-group">
<label for="username">用户名:</label>
<input type="text" id="username" name="username"
value="{{ form_data.get('username', [''])[0] }}">
{% if errors.get('username') %}<span class="error">{{ errors['username'][0] }}</span>{% end %}
</div>
<div class="form-group">
<label for="email">邮箱:</label>
<input type="email" id="email" name="email"
value="{{ form_data.get('email', [''])[0] }}">
{% if errors.get('email') %}<span class="error">{{ errors['email'][0] }}</span>{% end %}
</div>
<div class="form-group">
<label for="age">年龄:</label>
<input type="text" id="age" name="age"
value="{{ form_data.get('age', [''])[0] }}">
{% if errors.get('age') %}<span class="error">{{ errors['age'][0] }}</span>{% end %}
</div>
<button type="submit">注册</button>
</form>
</body>
</html>
""")
app_settings = {
'template_path': template_dir,
}
app = tornado.web.Application([
(r"/user_register", RegistrationHandler),
], **app_settings)
return app
if __name__ == "__main__":
validation_app = setup_validation_app()
validation_port = 8088
validation_app.listen(validation_port)
print(f"表单验证示例服务器已启动，请访问 http://localhost:{validation_port}/user_register")
tornado.ioloop.IOLoop.current().start()
