使用Flask构建RESTful API - TDD方式

2017-09-27

flask

Great things are done by a series of small things brought together – Vincent Van Gogh

本文的目的是提供一个易于遵循的基于项目的过程，介绍如何使用Flask框架创建RESTful API。

为什么使用 Flask?

有一点上下文 - 我在Django驱动的RESTful API上写了一堆文章。虽然Django爱好者是一个很好的资源，但并不是每个人都想在Django中编码。此外，熟悉其他框架总是很好。

学习Flask更容易和更快。这是非常容易设置和让事情运行。不像Django（它更重），你永远不会有功能，你没有使用。

我们所有的网络应用程序都是典型的，我们将使用TDD方法。真的很简单 测试驱动开发：

  • 写一个测试 - 测试将有助于在我们的应用程序中显示一些功能
  • 然后运行测试 - 测试应该失败，因为没有代码可以通过。
  • 编写代码 - 使测试通过
  • 运行测试 - 如果通过，我们确信我们编写的代码符合测试要求
  • 重构代码 - 删除重复，修剪大对象并使代码更易读。每次重构代码时重新运行测试
  • 重复 - 就是这样！

接着我们需要建立些什么？

我们将开发一个桶列表的API。桶列表是您想要实现的所有目标的列表，您想要实现的梦想，以及您在死亡之前想要体验的生活体验（或打斗）。因此，API应具备以下能力：

  • 创建一个桶单（通过给它一个名称/标题）
  • 检索现有的桶单
  • 更新它（通过更改它的名称/标题）
  • 删除现有的桶单
先决条件
  • Python3 - 一种可以让我们更快工作的编程语言（项目上线唯快不破）
  • Flask - 基于Werkzeug，Jinja 2的Python的微型开发框架
  • Virtualenv - 创建隔离虚拟环境的工具

我们从配置我们的Flask应用程序结构开始吧！

虚拟环境

首先，我们将创建我们的应用程序目录。在终端上，创建一个名为bucketlist的空目录mkdir bucketlist。然后cd 进入目录。创建一个孤立的虚拟环境：

$ virtualenv  venv

安装Autoenv 全局环境下使用pip install autoenv 以下是为什么 - Autoenv帮助我们设置将在每次cd进入我们的目录时运行的命令。它读取.env文件并为我们执行任何在那里。

创建一个.env文件并添加以下内容：

source env/bin/activate
export FLASK_APP="run.py"
export SECRET="some-very-long-string-of-random-characters-CHANGE-TO-YOUR-LIKING"
export APP_SETTINGS="development"
export DATABASE_URL="postgresql://localhost/flask_api"

第一行激活venv我们刚创建的虚拟环境。第2,3行出口我们的FLASK_APP, SECRET, APP_SETTINGS and DATABASE_URL变量。在开发过程中，我们将整合这些变量。

运行以下操作来更新和刷新.bashrc：

$ echo "source `which activate.sh`" >> ~/.bashrc
$ source ~/.bashrc

你会在终端上看到以下输出

虚拟环境

如果您安装了zsh， autoenv有时可能无法正常工作。一个很好的解决方法就是简单地执行设置好的 .env 文件

source .env

相反，如果您不想长期自动化东西，则不必使用autoenv。直接从终端出口的简单出口。

$ export FLASK_APP="run.py"
$ export APP_SETTINGS="development"
$ export SECRET="a-long-string-of-random-characters-CHANGE-TO-YOUR-LIKING"
$ export DATABASE_URL="postgresql://localhost/flask_api"

在我们的虚拟环境中，我们将创建一堆文件来布局我们的应用程序目录结构。这是它应该是什么样子：

├── bucketlist (this is the directory we cd into)
    ├── app
    │   ├── __init__.py
    │   └── models.py  
    ├── instance
    │   └── __init__.py
    ├── manage.py
    ├── requirements.txt
    ├── run.py
    └── test_bucketlist.py

完成以上操作后，使用pip安装Flask 。

(venv)$ pip install flask

环境配置

在应用程序启动之前，Flask需要一些可用的配置才能使用。由于环境（开发，生产或测试）需要特定的设置进行配置，我们就必须设置特定环境的东西，如一个secret key，debug mode并test mode在我们的配置文件。

如果您尚未创建一个目录，并调用它的实例。在这个目录里，创建一个名为config.py并初始化的.py。在我们的配置文件中，我们将添加以下代码：

# /instance/config.py

import os

class Config(object):
    """Parent configuration class."""
    DEBUG = False
    CSRF_ENABLED = True
    SECRET = os.getenv('SECRET')
    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')

class DevelopmentConfig(Config):
    """Configurations for Development."""
    DEBUG = True

class TestingConfig(Config):
    """Configurations for Testing, with a separate test database."""
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'postgresql://localhost/test_db'
    DEBUG = True

class StagingConfig(Config):
    """Configurations for Staging."""
    DEBUG = True

class ProductionConfig(Config):
    """Configurations for Production."""
    DEBUG = False
    TESTING = False

app_config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'staging': StagingConfig,
    'production': ProductionConfig,
}

Config类包含我们希望所有的环境中，默认情况下有常规设置。其他环境类继承自己，可以用于设置仅对它们唯一的设置。此外，该字典app_config用于导出我们指定的4个环境。这样做很方便，以便将来可以在其名称标签下导入配置。

需要注意的几个配置变量：

  • SECRET_KEY - 是用于生成哈希值的随机字符串，用于保护应用程序中的各种内容。它应该永远是公开的，以防止恶意攻击者访问它。
  • DEBUG - 告诉应用程序设置为在调试模式下运行True，允许我们使用Flask调试器。它也会在应用程序更新时自动重新加载。但是，它应该False在生产中设置。

配置数据库

安装要求

我们的数据库启动和运行所需的工具有：

我们可能使用了易于安装的数据库，如SQLite。但是，由于我们想要学习新的，强大的和令人敬畏的东西，所以我们将使用PostgreSQL。

SQLAlchemy是我们的对象关系映射器（ORM）。为什么要使用ORM，你问？ORM将原始SQL数据（称为查询）转换为数据，我们可以在调用的过程中理解称为对象的数据serialization，反之亦然（反序列化）。为了不复杂的原始SQL查询编写代码，为什么不使用为此而开发的测试工具？

我们按如下所示安装要求：

(venv)$ pip install flask-sqlalchemy psycopg2 flask-migrate

确保您已经在计算机上安装了PostgresSQL，并且它的服务器正在本地运行 port 5432

在您的终端中，创建Postgres数据库：

(venv) $ createdb test_db
(venv) $ createdb flask_api

Createdb是SQL命令的包装CREATE DATABASE。我们创

  • 测试数据库test_db为我们的测试环境。
  • 主要数据库flask_api用于开发环境。

我们使用了两个数据库，以便在运行测试时不会干扰主数据库的完整性。

创建应用程序

现在是需要写一些代码！由于我们正在创建一个API，所以我们将安装Flask-API扩展。

(venv)$ pip install Flask-API

Flask API是Django REST框架提供的相同的Web可浏览API的实现。这将有助于我们实现我们自己的可浏览的API。

在我们的空app/init.py文件中，我们将添加以下内容：

# app/__init__.py

from flask_api import FlaskAPI
from flask_sqlalchemy import SQLAlchemy

# local import
from instance.config import app_config

# initialize sql-alchemy
db = SQLAlchemy()

def create_app(config_name):
    app = FlaskAPI(__name__, instance_relative_config=True)
    app.config.from_object(app_config[config_name])
    app.config.from_pyfile('config.py')
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    db.init_app(app)

    return app

该create_app函数包装了一个新的Flask对象的创建，并在使用app.configDB并使用配置设置加载后返回db.init_app(app)。

我们还禁用了对SQLAlchemy的跟踪修改，因为它将来会被不推荐使用，因为它是显着的性能开销。对于调试爱好者，您True现在可以设置它。

现在，我们需要定义一个入门点来启动我们的应用程序。我们来编辑run.py文件。

import os

from app import create_app

config_name = os.getenv('APP_SETTINGS') # config_name = "development"
app = create_app(config_name)

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

运行！

现在我们可以在我们的终端上运行应用程序，看它是否可行：

(venv)$   flask run

我们也可以运行它python run.py。我们应该看到这样的输出：

run flask

数据模型

现在是时候创建我们的bucketlist模型了。模型是数据库中表的表示形式。在空models.py文件中添加以下内容：

# app/models.py

from app import db

class Bucketlist(db.Model):
    """This class represents the bucketlist table."""

    __tablename__ = 'bucketlists'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(255))
    date_created = db.Column(db.DateTime, default=db.func.current_timestamp())
    date_modified = db.Column(
        db.DateTime, default=db.func.current_timestamp(),
        onupdate=db.func.current_timestamp())

    def __init__(self, name):
        """initialize with name."""
        self.name = name

    def save(self):
        db.session.add(self)
        db.session.commit()

    @staticmethod
    def get_all():
        return Bucketlist.query.all()

    def delete(self):
        db.session.delete(self)
        db.session.commit()

    def __repr__(self):
        return "<Bucketlist: {}>".format(self.name)

这是我们在models.py文件中所写的发放：

  • 我们db从... 导入我们的连接app/init.py。
  • 接下来，我们创建了一个Bucketlist继承db.Model和分配表的类。名字bucketlists（应该总是复数）。因此，我们创建了一个表来存储我们的分类列表。
  • 该id字段包含主键，该name字段将存储桶列表的名称。
  • 该__repr__方法表示模型的对象实例，只要它是查询。
  • 该get_all()方法是一种静态方法，用于在单个查询中获取所有的分类列表。
  • 该save()方法将用于向DB添加新的存储区列表。
  • 该delete()方法将用于从DB中删除现有的存储列表。

迁移

迁移是将我们对模型所做的更改（如添加字段，删除模型等）传播到数据库模式中的一种方式。现在我们有一个定义的模型，我们需要告诉数据库创建相关的模式。

Flask-Migrate使用Alembic为我们自动生成迁移。它将用于此目的。

迁移脚本

每当我们编辑我们的模型时，迁移脚本将方便地帮助我们制作和应用迁移。分离迁移任务是不错的做法，而不是将它们与我们的应用程序中的代码进行混合。

也就是说，我们将创建一个名为manage.py的新文件。

我们的目录结构应该如下所示：

├── bucketlist
    ├── app
    │   ├── __init__.py
    │   └── models.py  
    ├── instance
    │   ├── __init__.py
    │   └── config.py
    ├── manage.py
    ├── requirements.txt
    ├── run.py
    └── test_bucketlist.py

将以下代码添加到manage.py：

# manage.py

import os
from flask_script import Manager # class for handling a set of commands
from flask_migrate import Migrate, MigrateCommand
from app import db, create_app
from app import models

app = create_app(config_name=os.getenv('APP_SETTINGS'))
migrate = Migrate(app, db)
manager = Manager(app)

manager.add_command('db', MigrateCommand)

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

在Manager此类跟踪所有的命令，并处理它们是如何从命令行调用。MigrateCommand包含一组迁移命令。我们还导入了模型，以便脚本可以找到要迁移的模型。经理还添加了迁移命令，并强制执行它们db。

我们将运行迁移初始化，使用以下db init命令：

(venv)$   python manage.py db init

您会注意到一个新创建的文件夹称为迁移。这将保持运行迁移所需的设置。“迁移”中的内容是一个名为“versions”的文件夹，它们将在创建时包含迁移脚本。

接下来，我们将使用以下db migrate命令运行实际迁移：

(venv)$   python manage.py db migrate

  INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
  INFO  [alembic.runtime.migration] Will assume transactional DDL.
  INFO  [alembic.autogenerate.compare] Detected added table 'results'
    Generating /bucketlist/migrations/versions/63dba2060f71_.py
    ...done

您还会注意到，在您的版本文件夹中有一个迁移文件。该文件由基于模型的Alembic自动生成。

最后，我们将使用以下db upgrade命令将迁移应用于数据库：

(venv)$   python manage.py db upgrade

INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 536e84635828, empty message

我们的数据库现在更新了我们的bucketlists表。如果您跳转到psql提示中，则会显示如何确认表是否存在：

postgresql

测试时间！

在我们的测试目录中，我们来创建测试。创建失败的测试是TD的第一步（不好）。这些测试将有助于指导我们创建我们的功能。首先写测试可能看起来很困难，但一旦你练习就真的很容易。

在父目录中创建一个名为test的测试文件test_bucketlist.py。该文件将包含以下内容：

  • 测试用例类来容纳我们所有的API测试。
  • setUp（）方法初始化我们的应用程序，它是测试客户端，并在应用程序的上下文中创建我们的测试数据库。
  • tearDown（）方法在测试完成后，删除测试变量并删除我们的测试数据库。
  • 测试以测试我们的API是否可以创建，读取，更新和删除存储列表。
# test_bucketlist.py
import unittest
import os
import json
from app import create_app, db

class BucketlistTestCase(unittest.TestCase):
    """This class represents the bucketlist test case"""

    def setUp(self):
        """Define test variables and initialize app."""
        self.app = create_app(config_name="testing")
        self.client = self.app.test_client
        self.bucketlist = {'name': 'Go to Borabora for vacation'}

        # binds the app to the current context
        with self.app.app_context():
            # create all tables
            db.create_all()

    def test_bucketlist_creation(self):
        """Test API can create a bucketlist (POST request)"""
        res = self.client().post('/bucketlists/', data=self.bucketlist)
        self.assertEqual(res.status_code, 201)
        self.assertIn('Go to Borabora', str(res.data))

    def test_api_can_get_all_bucketlists(self):
        """Test API can get a bucketlist (GET request)."""
        res = self.client().post('/bucketlists/', data=self.bucketlist)
        self.assertEqual(res.status_code, 201)
        res = self.client().get('/bucketlists/')
        self.assertEqual(res.status_code, 200)
        self.assertIn('Go to Borabora', str(res.data))

    def test_api_can_get_bucketlist_by_id(self):
        """Test API can get a single bucketlist by using it's id."""
        rv = self.client().post('/bucketlists/', data=self.bucketlist)
        self.assertEqual(rv.status_code, 201)
        result_in_json = json.loads(rv.data.decode('utf-8').replace("'", "\""))
        result = self.client().get(
            '/bucketlists/{}'.format(result_in_json['id']))
        self.assertEqual(result.status_code, 200)
        self.assertIn('Go to Borabora', str(result.data))

    def test_bucketlist_can_be_edited(self):
        """Test API can edit an existing bucketlist. (PUT request)"""
        rv = self.client().post(
            '/bucketlists/',
            data={'name': 'Eat, pray and love'})
        self.assertEqual(rv.status_code, 201)
        rv = self.client().put(
            '/bucketlists/1',
            data={
                "name": "Dont just eat, but also pray and love :-)"
            })
        self.assertEqual(rv.status_code, 200)
        results = self.client().get('/bucketlists/1')
        self.assertIn('Dont just eat', str(results.data))

    def test_bucketlist_deletion(self):
        """Test API can delete an existing bucketlist. (DELETE request)."""
        rv = self.client().post(
            '/bucketlists/',
            data={'name': 'Eat, pray and love'})
        self.assertEqual(rv.status_code, 201)
        res = self.client().delete('/bucketlists/1')
        self.assertEqual(res.status_code, 200)
        # Test to see if it exists, should return a 404
        result = self.client().get('/bucketlists/1')
        self.assertEqual(result.status_code, 404)

    def tearDown(self):
        """teardown all initialized variables."""
        with self.app.app_context():
            # drop all tables
            db.session.remove()
            db.drop_all()

# Make the tests conveniently executable
if __name__ == "__main__":
    unittest.main()

一点测试说明。在test_bucketlist_creation(self)我们内部使用测试客户端发送POST请求到/bucketlists/url。获得返回值，并将其状态代码置为等于状态代码201(Created)。如果等于201，测试断言是正确的，使测试通过。最后，它检查返回的响应是否包含刚刚创建的桶列表的名称。这是用做self.assertIn(a, b)如果断言评估为真，测试通过。

现在我们将运行测试如下：

(venv)$   python test_bucketlist.py

所有测试都必须失败。现在不要害怕 这是很好的，因为我们没有功能来测试通过。现在是创建将使我们的测试通过的API功能的时候了。

API功能

我们的API应该处理四个HTTP请求

  • POST - 用于创建bucketlist
  • GET - 用于使用其ID和许多分类列表检索一个bucketlist
  • PUT - 用于更新给定其ID的桶列表
  • 删除 - 删除给定其ID的存储桶列表

让我们立即完成这项工作。在我们的app/init.py文件里面，我们将对它进行如下编辑：

# app/__init__.py

# existing import remains

from flask import request, jsonify, abort

def create_app(config_name):
    from api.models import Bucketlist

    #####################
    # existing code remains #
    #####################

    @app.route('/bucketlists/', methods=['POST', 'GET'])
    def bucketlists():
        if request.method == "POST":
            name = str(request.data.get('name', ''))
            if name:
                bucketlist = Bucketlist(name=name)
                bucketlist.save()
                response = jsonify({
                    'id': bucketlist.id,
                    'name': bucketlist.name,
                    'date_created': bucketlist.date_created,
                    'date_modified': bucketlist.date_modified
                })
                response.status_code = 201
                return response
        else:
            # GET
            bucketlists = Bucketlist.get_all()
            results = []

            for bucketlist in bucketlists:
                obj = {
                    'id': bucketlist.id,
                    'name': bucketlist.name,
                    'date_created': bucketlist.date_created,
                    'date_modified': bucketlist.date_modified
                }
                results.append(obj)
            response = jsonify(results)
            response.status_code = 200
            return response

    return app

我们已经导入

  • request 用于处理我们的请求。
  • jsonify 使用application / json mimetype将JSON输出转换为Response对象。
  • abort 这将提前使用HTTP错误代码中止请求。

我们还在方法中from api.models import Bucketlist立即添加了导入，create_app以便我们可以访问Bucketlist模型，同时防止循环导入的恐怖。Flask @app.route在新功能之上提供了一个装饰器，def bucketlists()它强制我们只接受GET和POST请求。我们的功能首先检查它接收的请求的类型。如果是POST，则通过name从请求中提取出来并使用save()我们在模型中定义的方法来保存它来创建一个桶列表。因此，它将作为JSON对象返回新创建的bucketlist。如果是GET请求，它会从桶列表中获取所有的分类列表，并将一个列表作为JSON对象返回。如果我们的表中没有桶列表，它将返回一个空的JSON对象{}。

现在来看看我们的新功能GET和POST功能是否使我们的测试通过。

运行我们的测试

运行测试如下：

(venv)$ python test_bucketlists.py

2分5测试应通过。我们现在已经成功处理了GET和POST请求。

此时，我们的API只能创建并获取所有的分类列表。它不能使用其桶单ID获得单个桶单。此外，它也不能编辑桶列表，也不能从数据库中删除。要完成它，我们要添加这些功能。

添加PUT和DELETE功能

在我们的app/init.py文件中，我们来编辑如下：

# app/__init__.py

# existing import remains

def create_app(config_name):

    #####################
    # existing code remains #
    #####################

    ###################################
    # The GET and POST code is here
    ###################################

    @app.route('/bucketlists/<int:id>', methods=['GET', 'PUT', 'DELETE'])
    def bucketlist_manipulation(id, **kwargs):
     # retrieve a buckelist using it's ID
        bucketlist = Bucketlist.query.filter_by(id=id).first()
        if not bucketlist:
            # Raise an HTTPException with a 404 not found status code
            abort(404)

        if request.method == 'DELETE':
            bucketlist.delete()
            return {
            "message": "bucketlist {} deleted successfully".format(bucketlist.id) 
         }, 200

        elif request.method == 'PUT':
            name = str(request.data.get('name', ''))
            bucketlist.name = name
            bucketlist.save()
            response = jsonify({
                'id': bucketlist.id,
                'name': bucketlist.name,
                'date_created': bucketlist.date_created,
                'date_modified': bucketlist.date_modified
            })
            response.status_code = 200
            return response
        else:
            # GET
            response = jsonify({
                'id': bucketlist.id,
                'name': bucketlist.name,
                'date_created': bucketlist.date_created,
                'date_modified': bucketlist.date_modified
            })
            response.status_code = 200
            return response

    return app

我们现在定义了一个新的函数def bucketlist_manipulation()，它使用一个修饰器来强制它只处理GET, PUT and DELETEHttp请求。我们查询数据库以使用我们要访问的给定桶列表的id进行过滤。如果没有桶列表，它将aborts返回一个404 Not Found状态。第二个if-elif-else代码块分别处理删除，更新或获取bucketlist。

运行测试

现在，我们期望所有的测试都通过。让我们来看看他们是否真的通过了。

(venv)$ python test_bucketlist.py

我们现在应该看看所有的测试通过。

测试结果

使用 postman 和 curl 进行测试

打开 postman 输入http://localhost:5000/bucketlists/并发送具有名称作为有效载荷的POST请求。我们应该得到如下回应：

test using postman

我们也可以用Curl来玩，看看它在我们的终端上如何工作：

test using curl

结论

我们已经介绍了如何使用Flask创建一个测试驱动的RESTful API。我们了解到如何配置我们的Flask环境，将模型挂起，将迁移应用到DB中，创建单元测试并通过重构代码进行测试。

在本系列的第2部分中，我们将学习如何在API上强制执行用户身份验证和授权，并继续关注单元测试

https://scotch.io/tutorials/build-a-restful-api-with-flask-the-tdd-way