TDD开发容器化的Python微服务应用(一)-莱卡云

在这个课程中，你将学习如何使用Docker快速创建开发环境、管理多个微服务，应用程序在本地运行后，您将学习怎样在生产环境部署应用。我们也会练习TDD(测试驱动开发)，在你的项目中测试先行，我们重点将放在服务端的单元测试、功能和集成测试以及端到端的测试上面，以确保整个系统按预期工作。

有人问我为什么这么长的文章不分拆成几篇文章啊？这样阅读起来也方便啊，然而在我自己学习的过程中，这种整个一篇文章把一件事情从头到尾讲清楚的形式是最好的，能给读者提供一种沉浸式的学习体验，阅读完整个文章后有种酣畅淋漓的感觉，所以我选择这种一篇文章的形式。

扫描下面的二维码(或微信搜索iEverything)添加我微信好友(注明python)，然后可以加入到我们的python讨论群里面共同学习

架构

flask-microservices-main – Docker Compose 文件、Nginx、Admin 脚本
flask-microservices-users – 管理用户和认证的Flask 应用
flask-microservices-client – 客户端，React 应用
flask-microservices-swagger – Swagger API 文档

第一部分

1. 介绍

项目中提到的代码都已经归档到github上了，根据课程章节打上了tag。本课程是根据RealPython的课程进行改编的，希望看原版的就不要呆在这里骂我了，不送。

在第一部分，你将能够学习使用Docker、 Flask、MySQL来创建RESTful API服务。

我们将采取一种实用的方法来进行测试驱动开发(TDD)

课程开始之前，你应该要熟悉下面的这些主题：

Docker – Get started with Docker
Docker Compose – Get started with Docker Compose
Flask – Flaskr TDD

该部分课程使用到的工具库：

Python v3.6.4
Flask v0.12.2
Flask-Script v2.0.6
Flask-SQLAlchemy v2.3.2
Flask-Testing v0.7.1
PyMySQL v0.8.0
Gunicorn v19.7.1
Nginx v1.13.0
Docker v17.11.0-ce
Docker Compose v1.14.0

本节课程结束，你能学到下面的知识点：

用Flask开发一个RESTful API 服务
实践测试驱动开发(TDD)
在本地用Docker和Docker Compose配置和运行服务
将代码挂载到一个容器中去
在容器中运行单元测试和集成测试
不同容器的服务间的交互
在Docker容器中运行Python和Flask应用
安装Flask、Nginx和Gunicorn

2. 开始

本节课程我们将初始化项目框架、定义第一个服务。
创建一个新的项目、安装Flask：

$ mkdir flask-microservices-users && cd flask-microservices-users
$ mkdir project
$ pyenv virtualenv 3.6.4 tdd3
$ pyenv local tdd3
(tdd3)$ pip install Flask==0.12.2

在project 目录下面添加__init__.py文件，配置第一个路由：

# project/__init__.py
from flask import Flask, jsonify

# 初始化app
app = Flask(__name__)


@app.route('/ping', methods=['GET'])
def ping_pong():
    return jsonify({
        'status': 'success',
        'message': 'pong!'
    })

然后，添加Flask-Script，该依赖包可以用来在命令行管理Flask app:

(tdd3)$ pip install Flask-Script==2.0.6

在项目根目录下面添加一个manage.py的文件：

# manage.py
from flask_script import Manager
from project import app

manager = Manager(app)

if __name__ == '__main__':
    manager.run()

这里我们创建了一个新的Manager实例，来处理所有的从命令行输入的命令。运行程序：

(tdd3)$ python manage.py runserver

运行成功后，我们可以在浏览器中打开页面http://localhost:5000/ping，你能看到如下的json 信息输出到页面：

{
  "message": "pong!",
  "status": "success"
}

然后我们关掉服务(Ctrl+C)，在project目录下面新增一个config.py的文件，内容如下：

 #project/config.py
class BaseConfig:
    """基础配置"""
    DEBUG = False
    TESTING = False

class DevelopmentConfig(BaseConfig):
    """开发环境配置"""
    DEBUG = True

class TestingConfig(BaseConfig):
    """测试环境配置"""
    DEBUG = True
    TESTING = True

class ProductionConfig(BaseConfig):
    """生产环境配置"""
    DEBUG = False

然后更新__init__.py文件，在初始化的时候配置开发环境：

from flask import Flask, jsonify

# 初始化app
app = Flask(__name__)
# 环境配置
app.config.from_object('project.config.DevelopmentConfig')

@app.route('/ping', methods=['GET'])
def ping_pong():
    return jsonify({
        'status': 'success',
        'message': 'pong!'
    })

然后重新运行程序，我们能在终端中看到debug模式的提示信息：

(tdd3)$ python manage.py runserver
    * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
    * Restarting with stat
    * Debugger is active!
    * Debugger PIN: 770-657-842

现在当你的代码有任何改动的时候，应用都会自动加载，不需要手动重启了。为了保证我们的项目依赖能实时同步，这里我们将依赖包添加到一个requirements.txt的文件中，我们只需要利用pip的freeze命令就可以轻松拿到我们项目的依赖列表了：

(tdd3)$ pip freeze > requirements.txt

然后看看requirements.txt的内容吧:

Flask==0.12.2
Flask-Script==2.0.6

最后，在项目根目录下面添加一个.gitignore的文件：

__pycache__
env

初始化git 仓库，然后提交代码吧~

3. Docker 容器

这节课让我们来容器化Flask应用吧……
首先你需要确保你的环境中已经安装了Docker和Docker Compose：

$ docker -v
Docker version 17.05.0-ce, build 89658be
$ docker-compose -v
docker-compose version 1.14.0, build unknown

在项目的根目录下面创建一个Dockerfile的文件：

FROM python:3.6.4

# 设置工作目录
RUN mkdir -p /usr/src/app
WORKDIR /usr/src/app

# 添加依赖（利用Docker 的缓存）
ADD ./requirements.txt /usr/src/app/requirements.txt

# 安装依赖
RUN pip install -r requirements.txt

# 添加应用
ADD . /usr/src/app

# 运行服务
CMD python manage.py runserver -h 0.0.0.0

然后同样的在根目录下面创建一个docker-compose.yml的文件：

version: '2.1'

services:
  users-service:
    container_name: users-service
    build: .
    volumes:
      - '.:/usr/src/app'
    ports:
      - 5001:5000  # 端口暴露（主机端口:容器端口）

执行上面的配置文件将根据Dockerfile创建一个叫users-service的容器。
> docker-compose.yml文件中的目录是相对路径

volumes是用来将代码挂载进容器内部的，对于开发环境你最好这么做，不然你每次更新代码后都需要重新构建容器才会生效，显然这是非常影响效率的。

注意Docker compose 的文件版本，我们这里用的2.1，这其实与你安装的docker-compose版本没有任何关系的，只与Docker版本有关系（通过上面的连接可以选择合适的文件版本），但是文件版本3.x和2.x还是有很多不同的地方，注意查看文档。

然后我们来构建镜像：

$ docker-compose build

在第一次构建的时候会花费一点时间，不过别担心，由于Docker会缓存第一次的构建结果，后续的构建都会快得多，构建完成后，启动容器：

$ docker-compose up -d

-d标记是用来让容器在后台执行的

然后在浏览器中打开http://127.0.0.1:5001/ping，确保你能看到和上面看到的JSON 文件信息。
接下来在docker-compose.yml文件中增加一个环境变量，用来加载开发环境的配置信息：

version: '2.1'

services:
  users-service:
    container_name: users-service
    build: .
    volumes:
    - '.:/usr/src/app'
    ports:
    - 5001:5000 # 端口暴露（主机端口:容器端口）
    environment:
    - APP_SETTINGS=project.config.DevelopmentConfig

然后更新proejct/__init__.py文件，从环境变量获取应用的配置信息：

import os
from flask import Flask, jsonify

# 初始化app
app = Flask(__name__)
# 环境配置
app_settings = os.getenv('APP_SETTINGS')
app.config.from_object(app_settings)

@app.route('/ping', methods=['GET'])
def ping_pong():
    return jsonify({
        'status': 'success',
        'message': 'pong!'
    })

然后更新容器：

$ docker-compose up -d

怎么知道我们上面的这种方法就配置成功了呢？我们在project/__init__.py文件下面打印下配置信息：

import sys
print(app.config, file=sys.stderr)

因为我们将代码挂载到容器里面的，所以直接保存就会自动加载，然后查看日志：

$ docker-compose logs -f users-service

从日志中可以看出DEBUG=True，TESTING=False，证明是生效的（记得删除日志代码哦~~~）

4. 数据库安装配置

这节课我们将来配置MySQL数据库，启动运行在另外一个容器中，然后把它link到users-service容器中……

增加Flask-SQLAlchemy和PyMySQL到requirementx.txt文件中：

Flask-SQLAlchemy==2.3.2
PyMySQL==0.8.0

当然要记得安装这些依赖包：

(tdd3)$ pip install -r requirements.txt

然后更新config.py文件，添加SQLAlchemy数据库声明：

# project/config.py
import os

class BaseConfig:
    """基础配置"""
    DEBUG = False
    TESTING = False
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class DevelopmentConfig(BaseConfig):
    """开发环境配置"""
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

class TestingConfig(BaseConfig):
    """测试环境配置"""
    DEBUG = True
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_TEST_URL')

class ProductionConfig(BaseConfig):
    """生产环境配置"""
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')

接下来更新__init__.py文件，创建一个SQLAlchemy实例然后定义数据模型：

# project/__init__.py
import os
import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy

# 初始化app
app = Flask(__name__)
# 环境配置
app_settings = os.getenv('APP_SETTINGS')
app.config.from_object(app_settings)

# 初始化数据库
db = SQLAlchemy(app)

class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(128), nullable=False)
    email = db.Column(db.String(128), nullable=False)
    active = db.Column(db.Boolean(), default=False, nullable=False)
    created_at = db.Column(db.DateTime, nullable=False)

    def __init__(self, username, email):
        self.username = username
        self.email = email
        self.created_at = datetime.datetime.utcnow()

@app.route('/ping', methods=['GET'])
def ping_pong():
    return jsonify({
        'status': 'success',
        'message': 'pong!'
    })

在project目录下增加一个db的文件夹，在下面新建一个create.sql的文件：

CREATE DATABASE users_prod;
CREATE DATABASE users_dev;
CREATE DATABASE users_test;

然后在db目录下面添加文件Dockerfile:

FROM mysql:5.6

# 初始化的时候运行create.sql脚本
ADD create.sql /docker-entrypoint-initdb.d

这里我们继承官方的mysql镜像，在docker-entrypoint-initdb.d目录下面增加一个SQL文件，容器在初始化的时候就会执行这个sql文件。

更新docker-compose.yml：

version: '2.1'

services:
  users-db:
    container_name: users-db
    build: ./project/db
    ports:
      - 3307:3306
    environment:
      - MYSQL_ROOT_PASSWORD=root321
    healthcheck:
      test: exit 0

  users-service:
    container_name: users-service
    build: ./
    volumes:
      - '.:/usr/src/app'
    ports:
      - 5001:5000 # 暴露端口 - 主机:容器
    environment:
      - APP_SETTINGS=project.config.DevelopmentConfig
      - DATABASE_URL=mysql+pymysql://root:root321@users-db:3306/users_dev
      - DATABASE_TEST_URL=mysql+pymysql://root:root321@users-db:3306/users_test
    depends_on:
      users-db:
        condition: service_healthy
    links:
      - users-db

启动后注入环境变量exit code被发送后容器成功运行起来，MySQL将被绑定在宿主机的3307和容器的3306端口上。注意DATABASE_URL路径中使用的mysql+pymysql，因为我们使用的是python3.6.x版本和pymysql驱动。
检测：

(tdd3)$ docker-compose up -d --build

然后更新manage.py，增加创建数据库的命令：

from flask_script import Manager
from project import app, db

manager = Manager(app)

@manager.command
def recreate_db():
    """重新创建数据表."""
    db.drop_all()
    db.create_all()
    db.session.commit()

if __name__ == '__main__':
    manager.run()

新添加了一个recrete_db的命令，我们可以在命令行中执行该命令来将model映射到数据中：

(tdd3)$ docker-compose run users-service python manage.py recreate_db

如果一切正常的话，上面的命令能够执行成功，然后我们来验证下数据库中是否有对应的数据表了：

(tdd3)$ docker exec -it users-db mysql -uroot -p
Enter password:

然后输入上面我们环境变量中设置的root321就登录到MySQL数据库中了。然后执行下面的系列命令查看数据库、查看数据表结构：

mysql> show databases;
+--------------------+
| Database           |
+--------------------+
| information_schema |
| mysql              |
| performance_schema |
| users_dev          |
| users_prod         |
| users_test         |
+--------------------+
6 rows in set (0.00 sec)

mysql> use users_dev;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A

Database changed
mysql> show tables;
+---------------------+
| Tables_in_users_dev |
+---------------------+
| users               |
+---------------------+
1 row in set (0.00 sec)

mysql> desc users;
+------------+--------------+------+-----+---------+----------------+
| Field      | Type         | Null | Key | Default | Extra          |
+------------+--------------+------+-----+---------+----------------+
| id         | int(11)      | NO   | PRI | NULL    | auto_increment |
| username   | varchar(128) | NO   |     | NULL    |                |
| email      | varchar(128) | NO   |     | NULL    |                |
| active     | tinyint(1)   | NO   |     | NULL    |                |
| created_at | datetime     | NO   |     | NULL    |                |
+------------+--------------+------+-----+---------+----------------+
5 rows in set (0.00 sec)

从上面的命令我们可以看到model和我们的数据表已经映射成功了。

5. 测试

按照我们平时的开发习惯，是不是接下来就应该开发业务代码了？当然这样是可以的，但是我们这里不这样做，我们让测试先行，让测试来驱动开发，这样有什么好处呢？TDD最重要的功能就是保障代码的正确性，能够迅速发现、定位bug。关于TDD更多的知识可以自行Google。提供一个篇IBM关于TDD介绍的文章：浅谈测试驱动开发（TDD）

首先新增测试用的依赖包：Flask-Testing到requirements.txt文件下面（记得用pip命令安装哦~）：

Flask-Testing==0.7.1

在project目录下面新建一个tests的目录，然后在该目录使用下面新增几个文件：

(tdd3)$ touch __init__.py base.py test_config.py test_users.py

如果你对Flask-Testing还不太熟悉，在编写测试文件之前，可以先简单查看下文档。

更新base.py文件：

# project/tests/base.py
from flask_testing import TestCase
from project import app, db

class BaseTestCase(TestCase):
    def create_app(self):
        app.config.from_object('project.config.TestingConfig')
        return app

    def setUp(self):
        db.create_all()
        db.session.commit()

    def tearDown(self):
        db.session.remove()
        db.drop_all()

首先，我们确保测试使用的数据库URI不是生产环境的，避免对线上环境产生任何影响。每次测试的时候创建数据表、完成后删除表，确保能够清理测试数据。

必须指定一个create_app 方法，并且返回flask app 实例，如果没有定义将会抛出NotImplementedError异常。

另外在tearDown方法中增加db.session.remove()方法，这样能够确保每次测试完成的时候能够将SQLAlchemy的session属性移除掉，在下一次测试的时候始终是一个新的session。

更新test_config.py文件：

from flask import current_app
from flask_testing import TestCase
from project import app

class TestDevelopmentConfig(TestCase):
    def create_app(self):
        app.config.from_object('project.config.DevelopmentConfig')
        return app

    def test_app_is_development(self):
        self.assertTrue(app.config['SECRET_KEY'] == 'secret')
        self.assertTrue(app.config['DEBUG'] is True)
        self.assertFalse(current_app is None)
        self.assertTrue(
            app.config['SQLALCHEMY_DATABASE_URI'] ==
            'mysql+pymysql://root:root321@users-db:3306/users_dev'
        )

class TestTestingConfig(TestCase):
    def create_app(self):
        app.config.from_object('project.config.TestingConfig')
        return app

    def test_app_is_testing(self):
        self.assertTrue(app.config['SECRET_KEY'] == 'secret')
        self.assertTrue(app.config['DEBUG'])
        self.assertTrue(app.config['TESTING'])
        self.assertFalse(app.config['PRESERVE_CONTEXT_ON_EXCEPTION'])
        self.assertTrue(
            app.config['SQLALCHEMY_DATABASE_URI'] ==
            'mysql+pymysql://root:root321@users-db:3306/users_test'
        )

class TestProductionConfig(TestCase):
    def create_app(self):
        app.config.from_object('project.config.ProductionConfig')
        return app

    def test_app_is_production(self):
        self.assertTrue(app.config['SECRET_KEY'] == 'secret')
        self.assertFalse(app.config['DEBUG'])
        self.assertFalse(app.config['TESTING'])

更新test_users.py文件：

import json
from project.tests.base import BaseTestCase

class TestUserService(BaseTestCase):
    def test_users(self):
        """确保ping的服务正常."""
        response = self.client.get('/ping')
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 200)
        self.assertIn('pong', data['message'])
        self.assertIn('success', data['status'])

然后我们在manage.py中新增一个测试的命令行支持：

import unittest
from flask_script import Manager
from project import app, db

manager = Manager(app)

@manager.command
def recreate_db():
    """重新创建数据表."""
    db.drop_all()
    db.create_all()
    db.session.commit()

@manager.command
def test():
    """运行测试."""
    tests = unittest.TestLoader().discover('project/tests', pattern='test_*.py')
    result = unittest.TextTestRunner(verbosity=2).run(tests)
    if result.wasSuccessful():
        return 0
    return 1

if __name__ == '__main__':
    manager.run()

unittest支持非常简单的测试发现，为了兼容测试发现，所有的测试文件(test_xxx.py)必须是可以从项目的顶级目录导入的包或者模块，我们这里就是利用TestLoader.discover()方法去发现project/tests包下面的所有的test_xxx.py测试文件。然后使用TextTestRunner用来执行测试用例，对测试进行编排并把结果返回给用户。

如果是单个的测试文件，推荐你将所有的测试方法放在同一个文件中，这样能够方便的使用unittest.main()方法执行：

import unittest
import flask_testing

# TODO your test cases

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

然后可以执行命令python test_xxx.py进行测试。

由于我们这里有test_config.py和test_users.py两个业务不一样的测试用例，所以我们使用test runner来收集所有的测试结果并生成最后的测试报告。

由于我们更新了依赖包，所以需要重新构建镜像：

(tdd3)$ docker-compose up -d --build

然后运行测试命令：

(tdd3)$ docker-compose run users-service python manage.py test

然后我们可以看到类似于下面的测试没有通过的提示信息：

======================================================================
FAIL: test_app_is_development (test_config.TestDevelopmentConfig)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/project/tests/test_config.py", line 13, in test_app_is_development
    self.assertTrue(app.config['SECRET_KEY'] == 'secret')
AssertionError: False is not true

这是因为我们的app.config中还没有SECRET_KEY这个key，所以assertTrue肯定是不会通过的，然后我们在config.py的BaseConfig类中新增SECRET_KEY属性：

class BaseConfig:
    """基础配置"""
    DEBUG = False
    TESTING = False
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SECRET_KEY = 'secret'

重新执行测试命令：

(tdd3)$ docker-compose run users-service python manage.py test
Starting users-db ... done
test_app_is_development (test_config.TestDevelopmentConfig) ... ok
test_app_is_production (test_config.TestProductionConfig) ... ok
test_app_is_testing (test_config.TestTestingConfig) ... ok
test_users (test_users.TestUserService)
确保ping的服务正常. ... ok

----------------------------------------------------------------------
Ran 4 tests in 0.124s

OK

测试通过~~~

6. Flask Blueprint(蓝图)

上节课完成了我们的基本测试，这节课我们来用Blueprint(蓝图)来对项目进行重构。

还不了解Blueprint？可以先查看flask blueprint 文档。简单来说，Flask Blueprint提供了模块化管理程序路由的功能，使程序结构清晰、简单易懂。还是不太明白？没关系，这节课完成后我相信你一定会明白的~

在project目录下面新增api的目录，然后同样的需要在该目录下面新建几个文件：

(tdd3)$ touch __init__.py models.py views.py

然后更新views.py文件：

# project/api/views.py
from flask import Blueprint, jsonify

users_blueprint = Blueprint('users', __name__)

@users_blueprint.route('/ping', methods=['GET'])
def ping_pong():
    return jsonify({
        'status': 'success',
        'message': 'pong!'
    })

我们创建了一个users_blueprint的Blueprint实例，然后将该实例绑定到了ping_pong()方法上，这有什么用？继续往下看……

然后更新models.py文件：

# project/api/models.py
import datetime
from project import db

class User(db.Model):
    __tablename__ = "users"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(128), nullable=False)
    email = db.Column(db.String(128), nullable=False)
    active = db.Column(db.Boolean(), default=False, nullable=False)
    created_at = db.Column(db.DateTime, nullable=False)

    def __init__(self, username, email):
        self.username = username
        self.email = email
        self.created_at = datetime.datetime.utcnow()

我们可以看到上面的内容和project/__init__.py文件中的User类是一模一样的，没错，我们只是将这个地方代码分拆了而已，下面继续更新project/__init__.py文件：

# project/__init__.py
import os
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy

# 初始化数据库
db = SQLAlchemy()

def create_app():
    # 初始化应用
    app = Flask(__name__)
    # 环境配置
    app_settings = os.getenv('APP_SETTINGS')
    app.config.from_object(app_settings)
    # 安装扩展
    db.init_app(app)
    # 注册blueprint
    from project.api.views import users_blueprint
    app.register_blueprint(users_blueprint)
    return app

注意这里我们将实例化app的工作提取到了create_app的方法里面去，这是因为users_blueprint里面需要引用到当前文件下面的db实例，如果不把app放置到方法中去的话就会造成循环引用，什么意思？简单来说就是你中有我，我中有你，这对于程序来说是无法做出判断的~

接下来我们需要把所有其他文件引用project下面的app 的实例的都要替换掉(包括测试文件)：

from project import create_app

app = create_app()

更改完成以后我们重新构建镜像、创建数据库，最重要的是什么？测试：

(tdd3)$ docker-compose up -d
users-db is up-to-date
Starting users-service ...
Starting users-service ... done
(tdd3)$ docker-compose run users-service python manage.py recreate_db
Starting users-db ... done
(tdd3)$ docker-compose run users-service python manage.py test
Starting users-db ... done
test_app_is_development (test_config.TestDevelopmentConfig) ... ok
test_app_is_production (test_config.TestProductionConfig) ... ok
test_app_is_testing (test_config.TestTestingConfig) ... ok
test_users (test_users.TestUserService)
确保ping的服务正常. ... ok

----------------------------------------------------------------------
Ran 4 tests in 0.086s

OK

一切正常~~~

接下来我们根据RESTful的最佳实践利用TDD增加3个路由：

Endpoint	HTTP Method	CRUD Method	Result
/users	GET	查询	获取所有用户
/users/:id	GET	查询	获取单个用户
/users	POST	新增	新增用户

首先，在project/tests/test_users.py文件的TestUserService类中新增一个测试新增用户的方法：

def test_add_user(self):
    """确保能够正确添加一个用户的用户到数据库中"""
    with self.client:
        response = self.client.post(
            '/users',
            data=json.dumps(dict(username='cnych', email='qikqiak@gmail.com')),
            content_type='application/json',
        )
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 201)
        self.assertIn('qikqiak@gmail.com was added', data['message'])
        self.assertEqual('success', data['status'])

我们现在来执行测试肯定是不会通过的，因为路由/users还没实现的，所以接着我们在project/api/views.py中新增一个/users的处理方法：

# 注意要导入request
from flask import Blueprint, jsonify, request, render_template

@users_blueprint.route('/users', methods=['POST'])
def add_user():
    # 获取POST的数据
    post_data = request.get_json()
    email = post_data.get('email')
    user = User(username=post_data.get('username'), email=email)
    db.session.add(user)
    db.session.commit()
    response_data = {
        'status': 'success',
        'message': '%s was added!' % email
    }
    return jsonify(response_data), 201

注意上面我们add_user方法最终返回的数据，是从我们上面设计的测试代码中来的，这就是所谓的测试驱动我们的开发~

然后执行测试：

(tdd3)$ docker-compose run users-service python manage.py test
Starting users-db ... done
test_app_is_development (test_config.TestDevelopmentConfig) ... ok
test_app_is_production (test_config.TestProductionConfig) ... ok
test_app_is_testing (test_config.TestTestingConfig) ... ok
test_add_user (test_users.TestUserService)
确保能够正确添加一个用户的用户到数据库中 ... ok
test_users (test_users.TestUserService)
确保ping的服务正常. ... ok

----------------------------------------------------------------------
Ran 5 tests in 0.157s

OK

测试通过~~~

但是还没完呢？现在我们的代码还不够健壮，如果程序中出现了错误或者异常该怎么办呢？比如：

POST 的数据为空
POST 的数据无效 – 比如JSON 对象是空的或者包含一个错误的key
如果添加的用户在数据中已经存在？

来对这些用例添加一些测试代码：

def test_add_user_invalid_json(self):
    """如果JSON对象为空，确保抛出一个错误。"""
    with self.client:
        response = self.client.post(
            '/users',
            data=json.dumps(dict()),
            content_type='application/json'
        )
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 400)
        self.assertIn('Invalid payload', data['message'])
        self.assertEqual('fail', data['status'])

def test_add_user_invalid_json_keys(self):
    """如果JSON对象中没有username或email，确保抛出一个错误。"""
    with self.client:
        response = self.client.post(
            '/users',
            data=json.dumps(dict(email='qikqiak@gmail.com')),
            content_type='application/json'
        )
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 400)
        self.assertIn('Invalid payload', data['message'])
        self.assertEqual('fail', data['status'])

    with self.client:
        response = self.client.post(
            '/users',
            data=json.dumps(dict(username='cnych')),
            content_type='application/json'
        )
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 400)
        self.assertIn('Invalid payload', data['message'])
        self.assertEqual('fail', data['status'])

def test_add_user_duplicate_user(self):
    """如果邮件已经存在确保抛出一个错误。"""
    with self.client:
        self.client.post(
            '/users',
            data=json.dumps(dict(
                username='cnych',
                email='qikqiak@gmail.com'
            )),
            content_type='application/json'
        )
        response = self.client.post(
            '/users',
            data=json.dumps(dict(
                username='cnych',
                email='qikqiak@gmail.com'
            )),
            content_type='application/json'
        )
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 400)
        self.assertIn('Sorry. That email already exists.', data['message'])
        self.assertEqual('fail', data['status'])

现在我们支持测试命令，肯定是不会通过的，因为还没更新handler 呢：

from sqlalchemy import exc

@users_blueprint.route('/users', methods=['POST'])
def add_user():
    # 获取POST的数据
    post_data = request.get_json()
    if not post_data:
        response_data = {
            'status': 'fail',
            'message': 'Invalid payload.'
        }
        return jsonify(response_data), 400
    email = post_data.get('email')
    username = post_data.get('username')
    try:
        user = User.query.filter_by(email=email).first()
        if not user:
            # 证明数据库中不存在该email的用户，可以添加
            db.session.add(User(username=username, email=email))
            db.session.commit()
            response_data = {
                'status': 'success',
                'message': '%s was added!' % email
            }
            return jsonify(response_data), 201
        # 证明该email已经存在
        response_data = {
            'status': 'fail',
            'message': 'Sorry. That email already exists.'
        }
        return jsonify(response_data), 400
    except exc.IntegrityError as e:
        db.session.rollback()  # 出现异常了，回滚
        response_data = {
            'status': 'fail',
            'message': 'Invalid payload.'
        }
        return jsonify(response_data), 400

然后执行我们的测试命令，现在就能够测试通过了，如果出现了问题那么你应该仔细看看你的代码了~

接下来处理另外两个请求。

获取单个用户信息，还是先进行测试：

from project.api.models import User
from project import db

def test_get_user(self):
    user = User(username='cnych', email='qikqiak@gmail.com')
    db.session.add(user)
    db.session.commit()
    with self.client:
        response = self.client.get('/users/%d' % user.id)
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 200)
        self.assertTrue('created_at' in data['data'])
        self.assertEqual('cnych', data['data']['username'])
        self.assertEqual('qikqiak@gmail.com', data['data']['email'])
        self.assertEqual('success', data['status'])

然后来编写获取单个用户请求的处理函数，更新project/api/views.py文件：

@users_blueprint.route('/users/', methods=['GET'])
def get_user(user_id):
    """获取某用户的详细信息"""
    user = User.query.filter_by(id=user_id).first()
    response_object = {
        'status': 'success',
        'data': {
            'username': user.username,
            'email': user.email,
            'created_at': user.created_at
        }
    }
    return jsonify(response_object), 200

现在执行测试命令，测试能够通过了，那应该有哪一些错误处理的场景呢：

没有提供id
id不存在

然后我们来针对上面两种场景添加测试代码：

def test_get_user_no_id(self):
    """如果没有id的时候抛出异常。"""
    with self.client:
        response = self.client.get('/users/xxx')
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 400)
        self.assertIn('Param id error', data['message'])
        self.assertEqual('fail', data['status'])

def test_get_user_incorrect_id(self):
    """如果ID不存在则要抛出异常"""
    with self.client:
        response = self.client.get('/users/-1')
        data = json.loads(response.data.decode())
        self.assertEqual(response.status_code, 404)
        self.assertIn('User does not exist', data['message'])
        self.assertEqual('fail', data['status'])

然后根据上面我们的测试代码来更新get_user函数：

@users_blueprint.route('/users/', methods=['GET'])
def get_user(user_id):
    """获取某用户的详细信息"""
    response_object = {
        'status': 'fail',
        'message': 'User does not exist'
    }
    code = 404
    try:
        user = User.query.filter_by(id=int(user_id)).first()
        if user:
            response_object = {
                'status': 'success',
                'data': {
                    'username': user.username,
                    'email': user.email,
                    'created_at': user.created_at
                }
            }
            code = 200
    except ValueError:
        response_object = {
            'status': 'fail',
            'message': 'Param id error'
        }
        code = 400
    finally:
        return jsonify(response_object), code

然后继续执行我们的测试命令，通过~~~

然后是获取所有的用户列表的请求，这节就让我们的读者朋友自己来动手实践吧，最终代码我们会同步到github上去的，记住要用TDD的思想，先写测试代码，然后编写我们的网络请求函数，然后编写一些异常场景下面的测试代码，继续增强我们的请求函数，再测试。

上面的步骤完成后，我们试着在浏览器中打开http://127.0.0.1:5001/users接口，不出意外的话我们会看到如下的json信息输出：

{
    "data": {
        "users": []
    },
    "status": "success"
}

这是因为我们的数据库中还没有任何的数据，所以肯定这里得到的是一个空的数据列表。为了测试方便，我们在manage.py文件中增加一个命令来添加一些测试数据吧：

from project.api.models import User
@manager.command
def seed_db():
    """Seeds the database."""
    db.session.add(User(username='cnych', email="qikqiak@gmail.com"))
    db.session.add(User(username='chyang', email="icnych@gmail.com"))
    db.session.commit()

然后我们在命令行中执行seed_db命令：

(tdd3)$ docker-compose run users-service python manage.py seed_db

执行成功后，我们再次在浏览器中打开上面的接口，已经能够看到用户列表信息了。

7. 部署

上面的课程我们已经完成了测试和3个API接口的开发，现在我们来完成部署我们的应用。

首先在项目根目录新建一个docker-compose-prod.yml的文件，将docker-compose.yml文件的内容全部拷贝过来，然后去掉users-service下面的volumes，因为这是我们在开发阶段便于调试，将代码挂载到容器中的，生产环境就需要这样做了，然后就是需要将环境变量更改成生产环境的配置：

environment:
  - APP_SETTINGS=project.config.ProductionConfig
  - DATABASE_URL=mysql+pymysql://root:root321@users-db:3306/users_prod
  - DATABASE_TEST_URL=mysql+pymysql://root:root321@users-db:3306/users_test

然后执行构建镜像、创建数据库、添加初始数据、测试等命令：

(tdd3)$ docker-compose -f docker-compose-prod.yml up -d --build
(tdd3)$ docker-compose -f docker-compose-prod.yml run users-service python manage.py recreate_db
(tdd3)$ docker-compose -f docker-compose-prod.yml run users-service python manage.py seed_db
(tdd3)$ docker-compose -f docker-compose-prod.yml run users-service python manage.py test

测试阶段test_app_is_development这个测试用例肯定不会通过的，因为现在是生产环境了。

在生产环境我们一般使用Gunicorn来处理我们的网络请求，在requirements.txt文件中添加gunicorn==19.7.1，然后安装。安装完成后，在docker-compose-prod.yml文件中的users-service区域增加一条命令：

command: gunicorn -b 0.0.0.0:5000 manage:app

command会覆盖上面Dockerfile中的CMD命令。重新构建镜像：

(tdd3)$ docker-compose -f docker-compose-prod.yml up -d --build

由于增加了新的依赖文件，所以我们需要重新构建，--build是必须的。

接下来我们来安装Nginx，对我们的网络请求进行反向代理，提高我们接口的性能。在项目根目录下面新建一个nginx的文件夹，然后在该目录中新增Dockerfile文件：

FROM nginx:1.13.0

RUN rm /etc/nginx/conf.d/default.conf
ADD /flask.conf /etc/nginx/conf.d

然后在当前目录新建一个flask.conf的文件：

server {
    listen 80;
    location / {
        proxy_pass http://users-service:5000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

上面是一个普通的nginx配置文件，意思是从80端口来的请求都转发到后端的http://users-service:5000服务上，后端的users-service服务就是上面我们用gunicorn启动的服务，更多的nginx资料，可以Google查询。

接下来我们将我们的nginx服务添加到docker-compose-prod.yml文件中：

nginx:
  container_name: nginx
  build: ./nginx/
  restart: always
  ports:
    - 80:80
  depends_on:
    users-service:
      condition: service_started
  links:
    - users-service

然后将users-service区域中的端口暴露改成只暴露容器端口，因为主机不需要使用了：

expose:
  - '5000'

最后重新构建Docker镜像：

(tdd3)$ docker-compose -f docker-compose-prod.yml up -d --build

构建成功后，我们就可以通过你机器的IP 来访问我们的服务了，如果你有域名的话，就可以将你的域名解析到你主机的IP 上，然后就可以愉快的用域名进行访问我们的服务了，Done!

8. Jinja模板

接下来我们学习下服务端模板的使用，在project/api/views.py文件中增加一个新的路由处理函数：

from flask import Blueprint, jsonify, request, render_template
@users_blueprint.route('/', methods=['GET'])
def index():
    return render_template('index.html')

然后更新下Blueprint，增加对模板的配置：

users_blueprint = Blueprint('users', __name__, template_folder='./templates')

然后在project/api目录下增加一个templates的文件夹，在该文件夹下面添加一个index.html的文件：


html lang="en">
head>
    meta charset="UTF-8">
    title>Flask microservice app on Dockertitle>
    meta name="author" content="">
    meta name="description" content="">
    meta name="viewport" content="width=device-width,initial-scale=1">
    
    link href="//maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css" rel="stylesheet">
    {% block css %}{% endblock %}
head>
body>
div class="container">
  div class="row">
    div class="col-md-4">
      br>
      h1>All Usersh1>
      hr>br>
      form action="/" method="POST">
        div class="form-group">
          input name="username" class="form-control input-lg" type="text" placeholder="Enter a username" required>
        div>
        div class="form-group">
          input name="email" class="form-control input-lg" type="email" placeholder="Enter an email address" required>
        div>
        input type="submit" class="btn btn-primary btn-lg btn-block" value="Submit">
      form>
      br>
      hr>
      div>
        {% if users %}
          {% for user in users %}
            h4 class="well">strong>{{user.username}}strong> - em>{{user.created_at.strftime('%Y-%m-%d')}}em>h4>
          {% endfor %}
        {% else %}
          p>No users!p>
        {% endif %}
      div>
    div>
  div>
div>

script src="https://code.jquery.com/jquery-2.2.4.min.js">script>
{% block js %}{% endblock %}
body>
html>

然后再重新构建我们开发环境的镜像：

(tdd3)$ docker-compose -f docker-compose.yml up -d --build

构建完成后，在浏览器中打开http://127.0.0.1:5001

怎么来测试呢？如果当前数据库中没有任何数据：

def test_main_no_users(self):
    """没有用户"""
    response = self.client.get('/')
    self.assertEqual(response.status_code, 200)
    self.assertIn('No users!', response.data)

添加几条测试数据，获取所有用户列表，添加一条测试用例：

def test_main_with_users(self):
    """有多个用户的场景"""
    add_user('cnych', 'icnych@gmail.com')
    add_user('qikqiak', 'qikqiak@gmail.com')
    response = self.client.get('/')
    self.assertEqual(response.status_code, 200)
    self.assertIn(b'All Users', response.data)
    self.assertNotIn(b'No users!', response.data)
    self.assertIn(b'cnych', response.data)
    self.assertIn(b'qikqiak', response.data)

现在我们执行测试命令，会测试不通过的，因为还没有将用户列表的数据传递到首页，现在来更改project/api/views.py的index处理函数，增加用户列表的获取：

@users_blueprint.route('/', methods=['GET'])
def index():
    users = User.query.all()
    return render_template('index.html', users=users)

这样我们就能够将users列表传递到前端页面index.html中去了，然后可以通过标签{% for user in users %}对每一个用户进行渲染了，现在继续执行测试命令，通过~~~

在上面的前端页面中我们可以看到还有一个表单，用户点击提交按钮后可以添加一个新的用户到数据库中，当然继续我们的测试用例：

def test_main_add_user(self):
    """前端页面添加一个新的用户"""
    with self.client:
        response = self.client.post(
            '/',
            data=dict(username='cnych', email='cnych@gmail.com'),
            follow_redirects=True
        )
        self.assertEqual(response.status_code, 200)
        self.assertIn(b'All Users', response.data)
        self.assertNotIn(b'No users!', response.data)
        self.assertIn(b'cnych', response.data)

上面的测试代码中我们看到self.client.post()函数中多了一个follow_redirects=True参数，这样的话能确保post操作完成后页面能够重新刷新，这样的话首页的用户列表数据就能重新获取渲染了，现在继续执行我们的测试命令，会出现一个错误：

======================================================================
FAIL: test_main_add_user (test_users.TestUserService)
前端页面添加一个新的用户
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/src/app/project/tests/test_users.py", line 159, in test_main_add_user
    self.assertEqual(response.status_code, 200)
AssertionError: 405 != 200

----------------------------------------------------------------------

这是因为在首页的路由处理函数index中还没有对POST的请求进行处理，现在我们继续来更新project/api/views.py的index处理函数：

@users_blueprint.route('/', methods=['GET', 'POST'])
def index():
    if request.method == 'POST':
        username = request.form['username']
        email = request.form['email']
        db.session.add(User(username=username, email=email))
        db.session.commit()
    users = User.query.order_by(User.created_at.desc()).all()
    return render_template('index.html', users=users)

上面的函数针对POST的请求进行了数据库添加的操作，注意用户列表我们增加了一条根据created_at进行排序的规则还有methods增加了一个POST，现在再来执行我们的测试代码，通过~~~

上面的测试全部通过过后，现在我们将我们的代码部署到生产环境：

(tdd3)$ docker-compose -f docker-compose-prod.yml up -d --build

然后执行测试：

(tdd3)$ docker-compose -run users-service python manage.py test

然后在浏览器中打开http://127.0.0.1(因为生产环境是用Nginx做转发的，所以不用加端口)：

天下大事，合久必分，分久必合，下节课我们就来将项目进行拆分~~~

最后还是广告~~~