本文讲述如何使用 Tornado Web 框架来开发一个简单的 API 服务，以及如何使用 Docker 工具来构建镜像和部署服务。项目代码已在 GitHub 开源，JW Tornado Demo。

Tornado 介绍

Tornado 是一个 Python Web 框架，同时也是一个异步网络库。通过使用非阻塞网络 IO，它可以轻松处理上万连接，这使得它非常适合长轮询、WebSockets，以及需要为每一个用户维护一个连接的应用。

Tornado 还实现了 HTTP 协议的客户端（AsyncHTTPClient）和服务端（HTTPServer），可以用它来发起和接收 HTTP 请求。因为它提供了 HTTPServer，所以用它编写的 HTTP 服务（亦即 Web 应用）不再依赖其它应用容器（比如 Gunicorn）来运行，部署非常方便。

代码解读

目录结构

.
├── Dockerfile # Docker 镜像构建配置文件
├── LICENSE
├── Pipfile # Pipenv 配置文件
├── Pipfile.lock
├── README.md
├── docker-compose.yml # Docker Compose 配置文件
├── jwtornadodemo # 应用顶层 Python 包
│   ├── __init__.py
│   ├── account # 账号业务
│   │   ├── __init__.py
│   │   ├── cache.py
│   │   ├── const.py
│   │   ├── handler.py
│   │   ├── model.py
│   │   ├── service.py
│   │   └── vo.py
│   ├── app.py # 应用启动入口
│   ├── common # 公共模块
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── cache.py
│   │   ├── error.py
│   │   ├── handler.py
│   │   ├── model.py
│   │   └── vo.py
│   └── config # 应用配置
│       ├── __init__.py
│       ├── cache.py # Redis 缓存配置
│       ├── db.py # MongoDB 数据库配置
│       ├── env.py # 依赖环境的配置，提供了默认值，运行时可通过系统环境变量来覆盖
│       ├── logging.py # 日志配置
│       ├── session.py # Session 配置
│       └── tornado.py # Tornado 配置
├── static # 静态资源
│   └── README.md
└── tests # 单元测试
    ├── __init__.py
    └── test_config.py 

所有项目 Python 代码都放在 jwtornadodemo 这个顶级包下。

核心代码

编码规范严格遵循 PEP8。项目内模块之间引用均使用相对引用，以避免写死应用顶级包名。

app.py

app 模块为应用入口，定义了请求路由表，初始化日志，最后启动一个 Server 来接收网络请求。

import logging.config
import os

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

from . import config
from .account import handler as account_handler

app = tornado.web.Application([
    (r'/register', account_handler.RegisterUserHandler, None, 'register'),
    (r'/login', account_handler.LoginHandler, None, 'login'),
    (r'/isLogined', account_handler.IsLoginedHandler),
    (r'/logout', account_handler.LogoutHandler, None, 'logout'),
    (r'/account/edit', account_handler.EditUserHandler),
    (r'/account/info', account_handler.AccountInfoHandler),
], **config.TORNADO['settings'], session=config.SESSION)

if __name__ == '__main__':
    if not os.path.exists(config.PATH_LOG):
        os.makedirs(config.PATH_LOG)

    if not os.path.exists(config.PATH_UPLOAD):
        os.makedirs(config.PATH_UPLOAD)

    logging.config.dictConfig(config.LOGGING)

    server = tornado.httpserver.HTTPServer(app, xheaders=True)
    if config.DEBUG:
        server.listen(config.TORNADO['server']['port'])
    else:
        server.bind(config.TORNADO['server']['port'])
        server.start(config.TORNADO['server']['numprocs'])
    tornado.ioloop.IOLoop.current().start() 

config/env.py

config.env 模块将所有环境相关的配置抽取了出来集中管理，并且为每个配置项提供了默认值。部署到不同环境时可以通过设置系统环境变量来覆盖想要修改的配置项。这样不但代码做到了在各个环境通用，配置文件也做到了，这样就不用为每套环境维护一套配置文件。

import os

DEBUG = os.environ.get('DEBUG', 'true').lower() in ('true', 'yes', 'y', '1')

PATH_APP = os.environ.get('PATH_APP', os.path.normpath(
    os.path.join(os.path.dirname(__file__), '../..')))
PATH_DATA = os.environ.get('PATH_DATA', '/tmp')
PATH_LOG = os.path.join(PATH_DATA, 'log')
PATH_UPLOAD = os.path.join(PATH_DATA, 'upload')

LOGGING_LOGGER_LEVEL = os.environ.get('LOGGING_LOGGER_LEVEL', 'DEBUG')

TORNADO_SERVER_PORT = int(os.environ.get('TORNADO_SERVER_PORT', '8888'))
TORNADO_SERVER_NUMPROCS = int(os.environ.get('TORNADO_SERVER_NUMPROCS', '0'))

SESSION_COOKIE_SECRET = os.environ.get(
    'SESSION_COOKIE_SECRET', '4zi7D1)uw6VJ&Iz5@924y28Z@3@M3p!H')
SESSION_EXPIRES_SECONDS = int(os.environ.get('SESSION_EXPIRES_SECONDS',
                                             '86400'))

MONGODB_HOST = os.environ.get('MONGODB_HOST', 'localhost')
MONGODB_PORT = int(os.environ.get('MONGODB_PORT', '27017'))
MONGODB_NAME = os.environ.get('MONGODB_NAME', 'jw_tornado_demo')

REDIS_HOST = os.environ.get('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.environ.get('REDIS_PORT', '6379'))
REDIS_DB = int(os.environ.get('REDIS_DB', '0')) 

common/handler.py

common.handler 模块将各个业务模块 handler 的公共功能抽取了出来，里面实现了请求日志记录、会话管理、输出内容格式化等功能。

import json
import logging
import re
import traceback
from datetime import datetime

import torndsession.sessionhandler

from . import error as err
from . import cache, vo
from .. import config


class Handler(torndsession.sessionhandler.SessionBaseHandler):

    def prepare(self):
        super().prepare()

        logger = logging.getLogger('app')
        logger.debug('{} {} {} {}'.format(
            self.request.method, self.request.path,
            self.request.arguments, self.request.headers))

    def get_current_user(self):
        return self.session.get('user', None)

    def write_error(self, status_code, **kwargs):
        if self.settings.get('serve_traceback') and 'exc_info' in kwargs:
            message = traceback.format_exception(*kwargs['exc_info'])
        else:
            message = self._reason
        error = err.Error(message)

        return self.response_json(error, status_code)

    def response_json(self, error=None, status_code=200, **kwargs):
        data = {
            'code': err.ERROR_CODE_OK,
            'message': ''
        }
        if error:
            data['code'] = error.code
            data['message'] = (
                error.message if (config.DEBUG and error.message) else
                err.MESSAGES.get(error.code, '')
            )
        data.update(kwargs)

        ua = self.request.headers.get('User-Agent', '')
        if re.match(r'.+\s+MSIE\s+.+', ua):
            content_type = 'text/html; charset=utf-8'
        else:
            content_type = 'application/json; charset=utf-8'
        content = json.dumps(
            vo.jsonable(data),
            indent=(None if not config.DEBUG else 4),
            ensure_ascii=False)
        self.response(content, content_type, status_code)

    def response_html(self, template, error=None, status_code=200, **kwargs):
        data = {
            'code': err.ERROR_CODE_OK,
            'message': ''
        }
        if error:
            data['code'] = error.code
            data['message'] = (
                error.message if (config.DEBUG and error.message) else
                err.MESSAGES.get(error.code, '')
            )
        data.update(kwargs)

        content = self.render_string(template, **data)
        content_type = 'text/html; charset=utf-8'
        self.response(content, content_type, status_code)

    def response(self, content, content_type, status_code=200):
        self.set_status(status_code)
        self.set_header('Content-Type', content_type)
        self.finish(content) 

account/handler.py

account.handler 模块实现了帐号功能相关的接口，包括注册、登录、退出和信息查询等，每个接口都定义了一个 Form 来验证请求参数的合法性。

import json
import re

from wtforms.validators import InputRequired, Optional, AnyOf
from wtforms_tornado import Form
from pylib.form.field import StringField
from pylib.form.validator import DisplayWidth

from ..common import auth, error, handler
from .const import *
from .service import *
from .vo import *


class RegisterUserForm(Form):
    username = StringField(
        validators=[InputRequired(), DisplayWidth(3, 20)])
    password = StringField(
        validators=[InputRequired(), DisplayWidth(5, 20)])
    nickname = StringField(
        validators=[InputRequired(), DisplayWidth(2, 20)])
    gender = StringField(
        validators=[InputRequired(), AnyOf(USER_GENDER_CHOICES)])


class RegisterUserHandler(handler.Handler):

    def post(self):
        form = RegisterUserForm(self.request.arguments)
        if not form.validate():
            return self.response_json(
                error.Error(error.ERROR_CODE_PARAM_WRONG,
                            json.dumps(form.errors))
            )

        if not re.match(r'[a-z][a-z0-9_]{2,19}$', form.data['username'],
                        re.IGNORECASE):
            return self.response_json(error.Error(
                error.ERROR_CODE_PARAM_WRONG))

        user, error = register_user(**form.data)
        if error:
            return self.response_json(error)

        self.response_json(user=UserVO(user)(self))


class LoginForm(Form):
    username = StringField(
        validators=[InputRequired(), DisplayWidth(3, 50)])
    password = StringField(
        validators=[InputRequired(), DisplayWidth(5, 20)])


class LoginHandler(handler.Handler):

    def get(self):
        form = LoginForm(self.request.arguments)
        if not form.validate():
            return self.response_json(
                error.Error(error.ERROR_CODE_PARAM_WRONG,
                            json.dumps(form.errors))
            )

        user, error = verify_password(
            form.data['username'], form.data['password']
        )
        if error:
            return self.response_json(error)

        self.session['user'] = user

        self.response_json(user=UserVO(user)(self))


class IsLoginedHandler(handler.Handler):

    def get(self):
        user = self.session.get('user', None)

        self.response_json(user=UserVO(user)(self))


class LogoutHandler(handler.Handler):

    @auth.authenticated()
    def get(self):
        del self.session['user']

        self.response_json()


class EditUserForm(Form):
    nickname = StringField(
        validators=[Optional(), DisplayWidth(2, 20)])
    gender = StringField(
        validators=[Optional(), AnyOf(USER_GENDER_CHOICES)])


class EditUserHandler(handler.Handler):

    @auth.authenticated()
    def post(self):
        form = EditUserForm(self.request.arguments)
        if not form.validate():
            return self.response_json(
                error.Error(error.ERROR_CODE_PARAM_WRONG,
                            json.dumps(form.errors))
            )

        user, error = edit_user(self.current_user['_id'], form.data)
        if error:
            return self.response_json(error)

        self.session['user'] = user

        self.response_json(user=UserVO(user)(self))


class AccountInfoHandler(handler.Handler):

    @auth.authenticated()
    def get(self):
        user = user_info(self.current_user['_id'])

        self.response_json(user=UserVO(user)(self)) 

本地开发

IDE 推荐

IDE 推荐使用 Visual Studio Code，微软出品的一款支持多种语言的免费开发工具。其用法非常类似于 Sublime Text，熟悉 Sublime Text 的开发人员可以很快上手。相比于 Sublime Text，VSCode 的优势在于：一是微软出品，版本更新比较快，社区生态比较好；二是除了社区提供的大量插件，微软官方还提供了许多热门语言的插件，包括 Python；三是完全免费，终于可以摆脱 Sublime Text 时不时的购买提醒。关于如何使用 VS Code 来开发和调试 Python 程序，可参考本站的另一篇 博文。

安装 Python 3 和 Pipenv

Python 3 如今已是大多数 Python 应用开发的首选。在 Python 3 Readinees 网站上显示的 PyPI 下载排名前 360 的包里面，只有极少数还不支持 Python 3。这里面还有不少是一些工具类软件，比如 ansible、Fabric、supervisor，一般很少会在业务代码里用到。各个平台上如何安装 Python 请参考 Properly Installing Python，这里不再详述。

一直以来 Python 社区都缺少一款类似于 JavaScript NPM 这样的应用依赖包管理工具，直到 Pipenv 的出现。Pipenv 现在已经是 Python 官方推荐的打包工具，具体安装可参考 Pipenv & Virtual Environments。

运行应用

首先，从 GitHub 克隆代码到本地；

$ git clone git@github.com:jaggerwang/jw-tornado-demo.git
$ cd jw-tornado-demo 

其次，安装项目的所有 PyPI 依赖；

$ pipenv sync 

最后，运行应用；

$ pipenv run python -m jwtornadodemo.app 

这里是使用 pipenv 直接运行应用，也可以先激活虚拟环境，再在虚拟环境里运行应用。

$ pipenv shell
$ python -m jwtornadodemo.app 

注意，我们是使用模块方式来运行应用。这是因为在 app 模块里使用了相对引入（可以避免在项目代码里写死项目包名），而 Python 的相对引入要求模块文件不能作为入口脚本来运行（这种情况下模块的 __package__ 属性为空），所以只能使用模块方式来运行（这种情况下模块的 __package__ 属性为 -m 参数的值）。

单元测试

单元测试仅有一个示例，用来演示单元测试如何编写和运行。我们使用 Python 内置的 unittest 模块来编写和运行单元测试。unittest 为自动扫描检测项目内的所有单元测试文件来运行，默认匹配规则为 test*.py。

$ pipenv run python -m unittest 

API 测试

API 测试简单的场景可以通过命令行工具 curl 或者图形界面工具 Postman 来手工测试。复杂或者需要频繁测试的场景则最好编写自动化测试脚本来完成。

curl 命令用法如下（假设服务以默认端口在本地启动）。

$ curl http://localhost:8888/isLogined
{
    "code": 0,
    "message": ""
} 

Docker 部署

Docker 现在已经是应用部署的标准，Docker 对运维工作带来的改进是革命性的，甚至导致了许多运维人员的失业。因为 Docker 大大简化了运维工作，许多中小团队不再需要专门的运维人员，而由开发来兼任。Docker 安装请参考 官方安装指南。

构建应用镜像

克隆代码到本地，然后执行 docker build 命令来构建应用镜像。

$ git clone git@github.com:jaggerwang/jw-tornado-demo.git
$ cd jw-tornado-demo
$ docker build -t jw-tornado-demo . 

其中 jw-tornado-demo 为镜像 Tag，后面运行镜像的时候会用同样的 Tag 来指定哪个镜像，构建和运行镜像需要在同一台主机上。当然也可以把本地构建好的镜像 push 到某个镜像仓库，然后在部署目标主机上 pull 该镜像后就可以运行该镜像了。

用来告诉 Docker 如何构建镜像的 Dockerfile 如下：

FROM python:3

ENV APP_PATH=/app
ENV DATA_PATH=/data

WORKDIR $APP_PATH

COPY ./Pipfile* ./
RUN pip install pipenv
RUN pipenv sync

COPY . .

VOLUME $DATA_PATH

EXPOSE 8888

CMD pipenv run python -m jwtornadodemo.app 

FROM 指令表明我们用的基础镜像是 python，并且是版本 3 的最新版本。我们先通过 COPY 指令只把 Pipenv 的配置文件拷贝进镜像，然后安装应用依赖包，再拷贝剩余代码进镜像。这样做的好处是，只要 Pipenv 的配置文件不发生改动，则 Docker 会使用上次缓存的依赖包，无需重新下载和安装。这样能大大加快镜像构建速度，因为下载和安装依赖包是构建过程中最耗时的步骤。

运行镜像

使用 docker run 命令来在 Docker 容器里运行镜像，但我们的应用还依赖 MySQL 和 Redis 服务，需要一并启动。这种情况我们就需要用到 Docker Compose，它用来启动一组互相有依赖关系的服务。

在项目根目录下执行如下命令即可启动应用及其依赖的 MySQL 和 Reids 服务。

$ docker-compose up 

Docker Compose 配置文件如下：

version: "2"
services:
  app:
    image: jw-tornado-demo
    environment:
      DEBUG: 'false'
      PATH_APP: /app
      PATH_DATA: /data
      LOGGING_LOGGER_LEVEL: INFO
      TORNADO_SERVER_PORT: 8888
      TORNADO_SERVER_NUMPROCS: 0
      SESSION_COOKIE_SECRET: xxxxxxxx
      SESSION_EXPIRES_SECONDS: 86400
      MONGODB_HOST: mongodb
      MONGODB_PORT: 27017
      MONGODB_NAME: jw_tornado_demo
      REDIS_HOST: redis
      REDIS_PORT: 6379
      REDIS_DB: 0
    ports:
    - 19900:8888
    volumes:
    - ~/data/jw-tornado-demo/app:/data
  mongodb:
    image: mongo:3
    volumes:
    - ~/data/jw-tornado-demo/mongodb:/data/db
  redis:
    image: redis:4
    command:
    - redis-server
    - --appendonly
    - 'yes'
    volumes:
    - ~/data/jw-tornado-demo/redis:/data 

请根据个人环境修改配置文件。services 下有三个节点，表明会启动单个服务，分别是 app、mongodb 和 redis。每个 service 下的 image 指明了哪个镜像的哪个版本。environment 用来给容器里的应用传递环境变量，一些依赖环境的配置可以通过这种方式来传递。ports 将容器里应用监听的端口映射到主机上的某个端口，以便外部可以通过主机上的映射端口来访问容器内部的应用服务。volumes 将主机上的目录挂载到容器内的某个目录，这样容器内读写该目录就相当于是读写主机目录，并且在容器销毁后该目录内容会得到保留。

参考资料

1. JW Tornado Demo

本文转自 https://blog.jaggerwang.net/tornado-api-service-develop/，如有侵权，请联系删除。