跳至内容
概览
软件开发
Python开发
Python框架
Tornado
视图进阶与路由进阶

视图进阶与路由进阶

Tornado 的 RequestHandler 提供了一套完整的生命周期钩子，允许在请求处理的各个阶段插入逻辑。本文介绍路由进阶用法（别名与参数传递）、视图方法执行顺序、输出缓冲机制以及内置的用户认证支持。

路由进阶

url() 与 initialize

通过 tornado.web.url() 构造路由时，可传入 kwargs 参数字典。Tornado 会在实例化 Handler 后调用 initialize(**kwargs) 将这些参数注入视图：

from tornado.web import Application, url

app = Application([
    url(r"/admin", AdminHandler, {"title": "管理后台", "require_auth": True}),
])

class AdminHandler(tornado.web.RequestHandler):
    def initialize(self, title, require_auth=False):
        self.title = title
        self.require_auth = require_auth

    def get(self):
        self.write(f"欢迎来到 {self.title}")

initialize 的典型用途是注入公共依赖（数据库连接、配置项等），避免在每个方法中重复获取。

reverse_url 路由反解析

给路由命名后，可通过路由名称生成 URL，避免硬编码路径：

app = Application([
    url(r"/user/(\d+)", UserHandler, name="user_detail"),
])

在 Handler 中调用：

url = self.reverse_url("user_detail", 42)  # 生成 "/user/42"
self.redirect(url)

在模板中调用：

<a href="{{ reverse_url('user_detail', user.id) }}">查看用户</a>

视图生命周期

每个请求的处理过程会依次触发以下方法：

正常执行顺序 

set_default_headers()
  ↓
initialize(**kwargs)
  ↓
prepare()
  ↓
get() / post() / put() / delete() / ...（对应 HTTP 方法）
  ↓
on_finish()

抛出异常时的执行顺序

当 prepare() 或 HTTP 方法中抛出异常时，write_error() 替代正常方法执行：

set_default_headers()
  ↓
initialize(**kwargs)
  ↓
prepare()  ← 若此处抛出异常，跳转到 write_error
  ↓
get() / post() / ...  ← 若此处抛出异常，跳转到 write_error
  ↓
write_error(status_code, **kwargs)
  ↓
on_finish()

prepare

prepare() 在 HTTP 方法执行前调用，适合做通用的前置处理，如解析 JSON 请求体或校验公共参数：

import json

class BaseApiHandler(tornado.web.RequestHandler):
    def prepare(self):
        content_type = self.request.headers.get("Content-Type", "")
        if "application/json" in content_type:
            try:
                self.json_body = json.loads(self.request.body)
            except json.JSONDecodeError:
                self.send_error(400, reason="Invalid JSON")
        else:
            self.json_body = {}

prepare() 也可以是 async def，用于在处理请求前做异步操作（如查询数据库验证 Token）。

on_finish

on_finish() 在响应发送完毕后调用，适合做日志记录、资源清理等收尾工作：

import time

class TimedHandler(tornado.web.RequestHandler):
    def prepare(self):
        self._start_time = time.time()

    def on_finish(self):
        duration = time.time() - self._start_time
        print(f"{self.request.method} {self.request.uri} 耗时 {duration:.3f}s")
on_finish() 在响应已发送后执行，此时不能再调用 write() 或修改响应头。

write_error

覆盖 write_error() 可自定义错误响应格式：

class BaseHandler(tornado.web.RequestHandler):
    def write_error(self, status_code, **kwargs):
        self.set_header("Content-Type", "application/json")
        message = self._reason
        if "exc_info" in kwargs:
            # kwargs["exc_info"] 是 (type, value, traceback) 元组
            exc = kwargs["exc_info"][1]
            message = str(exc)
        self.write({"error": status_code, "message": message})

输出缓冲机制

Tornado 采用缓冲输出模型：write() 将内容追加到内存缓冲区，不立即发送；响应只有在 finish() 被调用时才真正发送到客户端。

class StreamHandler(tornado.web.RequestHandler):
    async def get(self):
        for i in range(5):
            self.write(f"chunk {i}\n")
            await self.flush()   # 立即冲刷缓冲区，发送已写内容
        # finish() 由框架在方法返回后自动调用

三个关键方法：

方法作用
write(chunk)将内容追加到缓冲区
flush()将缓冲区内容发送给客户端（不结束请求）
finish(chunk=None)发送缓冲区内容并关闭连接，请求结束

用户认证

@authenticated 装饰器

tornado.web.authenticated 是内置的认证装饰器。被装饰的方法会先调用 get_current_user()，若返回 None 或 False，则自动重定向到 settings["login_url"]：

import tornado.web

class ProfileHandler(tornado.web.RequestHandler):
    def get_current_user(self):
        return self.get_secure_cookie("user_id")

    @tornado.web.authenticated
    def get(self):
        user_id = self.current_user.decode()
        self.write(f"当前用户: {user_id}")

配套配置：

settings = {
    "cookie_secret": "your-secret-key",
    "login_url": "/login",   # 未认证时重定向到此 URL
}

登录与登出 

class LoginHandler(tornado.web.RequestHandler):
    def post(self):
        username = self.get_body_argument("username")
        password = self.get_body_argument("password")
        # 验证逻辑（查数据库等）
        if verify_user(username, password):
            self.set_secure_cookie("user_id", str(user.id))
            self.redirect("/")
        else:
            self.render("login.html", error="用户名或密码错误")

class LogoutHandler(tornado.web.RequestHandler):
    def get(self):
        self.clear_cookie("user_id")
        self.redirect("/login")

在 prepare 中统一鉴权

对于 API 接口，推荐在 prepare() 中统一校验 Token，而非逐个方法装饰：

class ApiBaseHandler(tornado.web.RequestHandler):
    async def prepare(self):
        token = self.request.headers.get("Authorization", "")
        if not token.startswith("Bearer "):
            self.set_status(401)
            self.write({"error": "Unauthorized"})
            self.finish()
            return
        # 可在此异步查询数据库验证 token
        self.current_user = await verify_token(token[7:])
最后更新于
请求与响应模板引擎