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

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

标签: python   flask   TDD   docker   react   kubernetes   Microservice  

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

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

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

目录

  1. 介绍
  2. 开始
  3. Docker 配置
  4. 数据库安装配置
  5. 测试
  6. Flask Buleprint
  7. 部署
  8. Jinja模板

架构

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

第一部分

1. 介绍

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

在第一部分,你将能够学习使用DockerFlaskMySQL来创建RESTful API服务。

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

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

  1. Docker - Get started with Docker
  2. Docker Compose - Get started with Docker Compose
  3. Flask - Flaskr TDD

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

  1. Python v3.6.4
  2. Flask v0.12.2
  3. Flask-Script v2.0.6
  4. Flask-SQLAlchemy v2.3.2
  5. Flask-Testing v0.7.1
  6. PyMySQL v0.8.0
  7. Gunicorn v19.7.1
  8. Nginx v1.13.0
  9. Docker v17.11.0-ce
  10. Docker Compose v1.14.0

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

  1. Flask开发一个RESTful API 服务
  2. 实践测试驱动开发(TDD)
  3. 在本地用DockerDocker Compose配置和运行服务
  4. 将代码挂载到一个容器中去
  5. 在容器中运行单元测试和集成测试
  6. 不同容器的服务间的交互
  7. Docker容器中运行PythonFlask应用
  8. 安装FlaskNginxGunicorn

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的文件中,我们只需要利用pipfreeze命令就可以轻松拿到我们项目的依赖列表了:

(tdd3)$ pip freeze > requirements.txt

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

Flask==0.12.2
Flask-Script==2.0.6

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

__pycache__
env

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

3. Docker 容器

这节课让我们来容器化Flask应用吧…… 首先你需要确保你的环境中已经安装了DockerDocker 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.x2.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

print-log 从日志中可以看出DEBUG=True,TESTING=False,证明是生效的(记得删除日志代码哦~~~)

4. 数据库安装配置

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

增加Flask-SQLAlchemyPyMySQLrequirementx.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-Testingrequirements.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()方法,这样能够确保每次测试完成的时候能够将SQLAlchemysession属性移除掉,在下一次测试的时候始终是一个新的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.pytest_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.pyBaseConfig类中新增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_blueprintBlueprint实例,然后将该实例绑定到了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

测试通过~~~

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

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

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

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/<user_id>', 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/<user_id>', 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的文件:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Flask microservice app on Docker</title>
    <meta name="author" content="">
    <meta name="description" content="">
    <meta name="viewport" content="width=device-width,initial-scale=1">
    <!-- styles -->
    <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 Users</h1>
      <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>
<!-- scripts -->
<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 flask-index

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

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.pyindex处理函数,增加用户列表的获取:

@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.pyindex处理函数:

@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做转发的,所以不用加端口): page-index

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

最后还是广告~~~

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

微信公众号

扫描下面的二维码关注我们的微信公众帐号,在微信公众帐号中回复◉加群◉即可加入到我们的 kubernetes 讨论群里面共同学习。

wechat-account-qrcode

「真诚赞赏,手留余香」

阳明

请我喝杯咖啡?

使用微信扫描二维码完成支付

相关文章