使用Flask构建RESTful API - TDD方式

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上强制执行用户身份验证和授权,并继续关注单元测试

 教程来源: scotch.io

评论 0