flask url构建_如何为生产构建构建Flask-RESTPlus Web服务

南宫嘉

2023-12-01

flask url构建

by Greg Obinna

由格雷格·奥比纳(Greg Obinna)

如何为生产构建构建Flask-RESTPlus Web服务 (How to structure a Flask-RESTPlus web service for production builds)

In this guide I’ll show you a step by step approach for structuring a Flask RESTPlus web application for testing, development and production environments. I will be using a Linux based OS (Ubuntu), but most of the steps can be replicated on Windows and Mac.

在本指南中，我将向您展示逐步构建用于测试，开发和生产环境的Flask RESTPlus Web应用程序的方法。我将使用基于Linux的操作系统(Ubuntu)，但是大多数步骤都可以在Windows和Mac上复制。

Before continuing with this guide, you should have a basic understanding of the Python programming language and the Flask micro framework. If you are not familiar with those, I recommend taking a look at an introductory article - How to use Python and Flask to build a web app.

在继续阅读本指南之前，您应该对Python编程语言和Flask微框架有基本的了解。如果您不熟悉这些内容，建议阅读介绍性文章- 如何使用Python和Flask构建Web应用程序。

本指南的结构 (How this guide is structured)

This guide is divided into the following parts:

本指南分为以下几部分：

特征 (Features)

We’ll be using the following features and extensions within our project.

我们将在项目中使用以下功能和扩展。

Flask-Bcrypt: A Flask extension that provides bcrypt hashing utilities for your application.
Flask-Bcrypt ： Flask扩展，为您的应用程序提供bcrypt哈希实用程序 。
Flask-Migrate: An extension that handles SQLAlchemy database migrations for Flask applications using Alembic. The database operations are made available through the Flask command-line interface or through the Flask-Script extension.
Flask-Migrate ： 一种扩展，用于使用Alembic处理Flask应用程序SQLAlchemy数据库迁移。 可通过Flask命令行界面或Flask-Script扩展名使用数据库操作。
Flask-SQLAlchemy: An extension for Flask that adds support for SQLAlchemy to your application.
烧瓶SQLAlchemy ： 延期瓶，增加了支持SQLAlchemy您的应用程序。
PyJWT: A Python library which allows you to encode and decode JSON Web Tokens (JWT). JWT is an open, industry-standard (RFC 7519) for representing claims securely between two parties.
PyJWT ： 一个Python库，允许您编码和解码JSON Web令牌(JWT)。 JWT是一种开放的行业标准( RFC 7519 )，用于在两方之间安全地表示索赔。
Flask-Script: An extension that provides support for writing external scripts in Flask and other command-line tasks that belong outside the web application itself.
Flask-Script ： 一种扩展，提供对使用Flask编写外部脚本以及Web应用程序本身之外的其他命令行任务的支持。
Namespaces (Blueprints)
命名空间 ( 蓝图 )
Flask-restplus
烧瓶剩余
UnitTest
单元测试

什么是Flask-RESTPlus？ (What is Flask-RESTPlus?)

Flask-RESTPlus is an extension for Flask that adds support for quickly building REST APIs. Flask-RESTPlus encourages best practices with minimal setup. It provides a coherent collection of decorators and tools to describe your API and expose its documentation properly (using Swagger).

Flask-RESTPlus是Flask的扩展，增加了对快速构建REST API的支持。 Flask-RESTPlus鼓励以最少的设置进行最佳实践。它提供了装饰器和工具的一致集合，以描述您的API并正确公开其文档(使用Swagger)。

安装与安装 (Setup and Installation)

Check if you have pip installed, by typing the command pip --version into the Terminal , then press Enter.

通过在Terminal中键入命令pip --version来检查是否已安装pip --version ，然后按Enter。

pip --version

If the terminal responds with the version number, this means that pip is installed, so go to the next step, otherwise install pip or using the Linux package manager, run the command below on the terminal and press enter. Choose either the Python 2.x OR 3.x version.

如果终端响应版本号，则表示已安装pip，请转到下一步，否则安装pip或使用Linux软件包管理器，在终端上运行以下命令，然后按Enter。选择Python 2.x或3.x版本。

Python 2.x
Python 2.x

sudo apt-get install python-pip

Python 3.x
Python 3.x

sudo apt-get install python3-pip

Set up virtual environment and virtual environment wrapper (you only need one of these, depending on the version installed above):

设置虚拟环境和虚拟环境包装器(您只需要其中之一，取决于上面安装的版本)：

sudo pip install virtualenv

sudo pip3 install virtualenvwrapper

Follow this link for a complete setup of virtual environment wrapper.

单击此链接可完整设置虚拟环境包装程序。

Create a new environment and activate it by executing the following command on the terminal:

通过在终端上执行以下命令来创建新环境并激活它：

mkproject name_of_your_project

项目设置与组织 (Project Setup and Organization)

I will be using a functional structure to organize the files of the project by what they do. In a functional structure, templates are grouped together in one directory, static files in another and views in a third.

我将使用功能结构按其功能来组织项目文件。在功能结构中，模板在一个目录中分组在一起，静态文件在另一个目录中分组，视图在第三目录中分组。

In the project directory, create a new package called app. Inside app, create two packages main and test. Your directory structure should look similar to the one below.

在项目目录中，创建一个名为app的新软件包。在app内部，创建两个包main和test 。您的目录结构应类似于以下结构。

.
├── app
│   ├── __init__.py
│   ├── main
│   │   └── __init__.py
│   └── test
│       └── __init__.py
└── requirements.txt

We are going to use a functional structure to modularize our application.Inside the main package, create three more packages namely: controller, service and model. The model package will contain all of our database models while the service package will contain all the business logic of our application and finally the controller package will contain all our application endpoints. The tree structure should now look as follows:

我们将使用功能结构来模块化我们的应用程序。在main包中，再创建三个程序包： controller ， service和model 。 model包将包含我们所有的数据库模型，而service包将包含我们应用程序的所有业务逻辑，最后controller包将包含我们所有的应用程序端点。现在，树结构应如下所示：

.
├── app
│   ├── __init__.py
│   ├── main
│   │   ├── controller
│   │   │   └── __init__.py
│   │   ├── __init__.py
│   │   ├── model
│   │   │   └── __init__.py
│   │   └── service
│   │       └── __init__.py
│   └── test
│       └── __init__.py
└── requirements.txt

Now lets install the required packages. Make sure the virtual environment you created is activated and run the following commands on the terminal:

现在让我们安装所需的软件包。确保已激活创建的虚拟环境，并在终端上运行以下命令：

pip install flask-bcrypt

pip install flask-restplus

pip install Flask-Migrate

pip install pyjwt

pip install Flask-Script

pip install flask_testing

Create or update the requirements.txt file by running the command:

通过运行以下命令来创建或更新requirements.txt文件：

pip freeze > requirements.txt

The generated requirements.txt file should look similar to the one below:

生成的requirements.txt文件应类似于以下内容：

alembic==0.9.8
aniso8601==3.0.0
bcrypt==3.1.4
cffi==1.11.5
click==6.7
Flask==0.12.2
Flask-Bcrypt==0.7.1
Flask-Migrate==2.1.1
flask-restplus==0.10.1
Flask-Script==2.0.6
Flask-SQLAlchemy==2.3.2
Flask-Testing==0.7.1
itsdangerous==0.24
Jinja2==2.10
jsonschema==2.6.0
Mako==1.0.7
MarkupSafe==1.0
pycparser==2.18
PyJWT==1.6.0
python-dateutil==2.7.0
python-editor==1.0.3
pytz==2018.3
six==1.11.0
SQLAlchemy==1.2.5
Werkzeug==0.14.1

配置设定 (Configuration Settings)

In the main package create a file called config.py with the following content:

在main软件包中，创建一个名为config.py的文件，其内容如下：

import os

# uncomment the line below for postgres database url from environment variable
# postgres_local_base = os.environ['DATABASE_URL']

basedir = os.path.abspath(os.path.dirname(__file__))

class Config:
    SECRET_KEY = os.getenv('SECRET_KEY', 'my_precious_secret_key')
    DEBUG = False


class DevelopmentConfig(Config):
    # uncomment the line below to use postgres
    # SQLALCHEMY_DATABASE_URI = postgres_local_base
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_main.db')
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class TestingConfig(Config):
    DEBUG = True
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_test.db')
    PRESERVE_CONTEXT_ON_EXCEPTION = False
    SQLALCHEMY_TRACK_MODIFICATIONS = False


class ProductionConfig(Config):
    DEBUG = False
    # uncomment the line below to use postgres
    # SQLALCHEMY_DATABASE_URI = postgres_local_base


config_by_name = dict(
    dev=DevelopmentConfig,
    test=TestingConfig,
    prod=ProductionConfig
)

key = Config.SECRET_KEY

The configuration file contains three environment setup classes which includes testing, development, and production.

配置文件包含三个环境设置类，其中包括testing ， development和production 。

We will be using the application factory pattern for creating our Flask object. This pattern is most useful for creating multiple instances of our application with different settings. This facilitates the ease at which we switch between our testing, development and production environment by calling the create_app function with the required parameter.

我们将使用应用程序工厂模式来创建Flask对象。此模式对于使用不同设置创建应用程序的多个实例最有用。通过使用必需的参数调用create_app函数，可以方便我们在测试，开发和生产环境之间进行切换。

In the __init__.py file inside the main package, enter the following lines of code:

在main软件包内的__init__.py文件中，输入以下代码行：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_bcrypt import Bcrypt

from .config import config_by_name

db = SQLAlchemy()
flask_bcrypt = Bcrypt()


def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config_by_name[config_name])
    db.init_app(app)
    flask_bcrypt.init_app(app)

    return app

烧瓶脚本 (Flask Script)

Now let’s create our application entry point. In the root directory of the project, create a file called manage.py with the following content:

现在，让我们创建应用程序入口点。在项目的根目录中，创建一个名为manage.py的文件，其内容如下：

import os
import unittest

from flask_migrate import Migrate, MigrateCommand
from flask_script import Manager

from app.main import create_app, db

app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')

app.app_context().push()

manager = Manager(app)

migrate = Migrate(app, db)

manager.add_command('db', MigrateCommand)

@manager.command
def run():
    app.run()

@manager.command
def test():
    """Runs the unit tests."""
    tests = unittest.TestLoader().discover('app/test', pattern='test*.py')
    result = unittest.TextTestRunner(verbosity=2).run(tests)
    if result.wasSuccessful():
        return 0
    return 1

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

The above code within manage.py does the following:

manage.py中的上述代码执行以下操作：

line 4 and 5 imports the migrate and manager modules respectively (we will be using the migrate command soon).
line 4和第5 line 4分别导入了migration和manager模块(我们将很快使用migration命令)。
line 9 calls the create_app function we created initially to create the application instance with the required parameter from the environment variable which can be either of the following - dev, prod, test. If none is set in the environment variable, the default dev is used.
line 9调用我们最初创建的create_app函数，以使用环境变量中的必需参数创建应用程序实例，该环境变量可以是以下任意一个dev ， prod ， test 。如果在环境变量中未设置任何值，则使用默认dev 。
line 13 and 15 instantiates the manager and migrate classes by passing the app instance to their respective constructors.
line 13和15 line 13通过将app实例传递给它们各自的构造函数来实例化管理器和迁移类。
In line 17,we pass the db and MigrateCommandinstances to the add_command interface of the managerto expose all the database migration commands through Flask-Script.
在line 17 ，我们将db和MigrateCommand实例传递给manager的add_command接口，以通过Flask-Script公开所有数据库迁移命令。
line 20 and 25 marks the two functions as executable from the command line.
line 20和25 line 20这两个功能标记为可从命令行执行。

Flask-Migrate exposes two classes, Migrate and MigrateCommand. The Migrateclass contains all the functionality of the extension. The MigrateCommand class is only used when it is desired to expose database migration commands through the Flask-Script extension.

Flask-Migrate公开了两个类Migrate和MigrateCommand 。 Migrate类包含扩展的所有功能。 仅当需要通过Flask-Script扩展公开数据库迁移命令时，才使用MigrateCommand类。

At this point, we can test the application by running the command below in the project root directory.

此时，我们可以通过在项目根目录中运行以下命令来测试应用程序。

python manage.py run

If everything is okay, you should see something like this:

如果一切正常，您应该会看到类似以下内容：

数据库模型和迁移 (Database Models and Migration)

Now let’s create our models. We will be using the db instance of the sqlalchemy to create our models.

现在让我们创建模型。我们将使用sqlalchemy的db实例创建模型。

The db instance contains all the functions and helpers from both sqlalchemyand sqlalchemy.orm and it provides a class called Model that is a declarative base which can be used to declare models.

db实例包含sqlalchemy和sqlalchemy.orm所有功能和帮助程序和它提供了一个称为Model的类，该类是可用于声明模型的声明性基础。

In the model package, create a file called user.py with the following content:

在model包中，创建一个名为user.py的文件，其内容如下：

from .. import db, flask_bcrypt

class User(db.Model):
    """ User Model for storing user related details """
    __tablename__ = "user"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    email = db.Column(db.String(255), unique=True, nullable=False)
    registered_on = db.Column(db.DateTime, nullable=False)
    admin = db.Column(db.Boolean, nullable=False, default=False)
    public_id = db.Column(db.String(100), unique=True)
    username = db.Column(db.String(50), unique=True)
    password_hash = db.Column(db.String(100))

    @property
    def password(self):
        raise AttributeError('password: write-only field')

    @password.setter
    def password(self, password):
        self.password_hash = flask_bcrypt.generate_password_hash(password).decode('utf-8')

    def check_password(self, password):
        return flask_bcrypt.check_password_hash(self.password_hash, password)

    def __repr__(self):
        return "<User '{}'>".format(self.username)

The above code within user.py does the following:

user.py中的上述代码执行以下操作：

line 3: The user class inherits from db.Model class which declares the class as a model for sqlalchemy.
line 3: user类继承自db.Model类，该类声明该类为sqlalchemy的模型。
line 7 through 13 creates the required columns for the user table.
line 7至13 user表创建所需的列。
line 21 is a setter for the field password_hash and it uses flask-bcryptto generate a hash using the provided password.
line 21是字段password_hash ，它使用flask-bcrypt使用提供的密码生成哈希。
line 24 compares a given password with already savedpassword_hash.
line 24给定的密码与已经保存的password_hash进行比较。

Now to generate the database table from the user model we just created, we will use migrateCommand through the manager interface. For managerto detect our models, we will have to import theuser model by adding below code to manage.py file:

现在要从我们刚刚创建的user模型生成数据库表，我们将通过manager界面使用migrateCommand 。为了让manager能够检测到我们的模型，我们将必须通过在manage.py文件中添加以下代码来导入user模型：

...
from app.main.model import user
...

Now we can proceed to perform the migration by running the following commands on the project root directory:

现在，我们可以通过在项目根目录上运行以下命令来继续执行迁移：

Initiate a migration folder using init command for alembic to perform the migrations.
使用init命令初始化一个迁移文件夹，以使Alembic能够执行迁移。

python manage.py db init

2. Create a migration script from the detected changes in the model using the migrate command. This doesn’t affect the database yet.

2.使用在模型中检测到的变化创建迁移脚本migrate命令。这还不影响数据库。

python manage.py db migrate --message 'initial database migration'

3. Apply the migration script to the database by using the upgrade command

3.使用upgrade命令将迁移脚本应用于数据库

python manage.py db upgrade

If everything runs successfully, you should have a new sqlLite database flask_boilerplate_main.db file generated inside the main package.

如果一切都成功运行，则应该在主软件包中生成一个新的sqlLite数据库flask_boilerplate_main.db文件。

Each time the database model changes, repeat the migrate and upgrade commands

每次数据库模型更改时，重复执行migrate和upgrade命令

测试中 (Testing)

组态 (Configuration)

To be sure the setup for our environment configuration is working, let’s write a couple of tests for it.

为确保环境配置的设置正常运行，让我们为它编写一些测试。

Create a file called test_config.py in the test package with the content below:

在测试包中创建一个名为test_config.py的文件，内容如下：

import os
import unittest

from flask import current_app
from flask_testing import TestCase

from manage import app
from app.main.config import basedir


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

    def test_app_is_development(self):
        self.assertFalse(app.config['SECRET_KEY'] is 'my_precious')
        self.assertTrue(app.config['DEBUG'] is True)
        self.assertFalse(current_app is None)
        self.assertTrue(
            app.config['SQLALCHEMY_DATABASE_URI'] == 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_main.db')
        )


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

    def test_app_is_testing(self):
        self.assertFalse(app.config['SECRET_KEY'] is 'my_precious')
        self.assertTrue(app.config['DEBUG'])
        self.assertTrue(
            app.config['SQLALCHEMY_DATABASE_URI'] == 'sqlite:///' + os.path.join(basedir, 'flask_boilerplate_test.db')
        )


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

    def test_app_is_production(self):
        self.assertTrue(app.config['DEBUG'] is False)


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

Run the test using the command below:

使用以下命令运行测试：

python manage.py test

You should get the following output:

您应该获得以下输出：

用户操作 (User Operations)

Now let’s work on the following user related operations:

现在，让我们进行以下与用户相关的操作：

creating a new user
创建一个新用户
getting a registered user with his public_id
使用其public_id获取注册用户
getting all registered users.
获取所有注册用户。

User Service class: This class handles all the logic relating to the user model.In the service package, create a new file user_service.py with the following content:

用户服务类：此类处理与用户模型有关的所有逻辑。在service包中，创建一个新文件user_service.py ，其内容如下：

import uuid
import datetime

from app.main import db
from app.main.model.user import User


def save_new_user(data):
    user = User.query.filter_by(email=data['email']).first()
    if not user:
        new_user = User(
            public_id=str(uuid.uuid4()),
            email=data['email'],
            username=data['username'],
            password=data['password'],
            registered_on=datetime.datetime.utcnow()
        )
        save_changes(new_user)
        response_object = {
            'status': 'success',
            'message': 'Successfully registered.'
        }
        return response_object, 201
    else:
        response_object = {
            'status': 'fail',
            'message': 'User already exists. Please Log in.',
        }
        return response_object, 409


def get_all_users():
    return User.query.all()


def get_a_user(public_id):
    return User.query.filter_by(public_id=public_id).first()


def save_changes(data):
    db.session.add(data)
    db.session.commit()

The above code within user_service.py does the following:

user_service.py中的上述代码执行以下操作：

line 8 through 29 creates a new user by first checking if the user already exists; it returns a success response_object if the user doesn’t exist else it returns an error code 409 and a failure response_object.
line 8至29 line 8通过首先检查该用户是否已存在来创建新用户；如果用户不存在，则返回成功response_object ，否则返回错误代码409和失败response_object 。
line 33 and 37 return a list of all registered users and a user object by providing the public_id respectively.
line 33和37 line 33分别通过提供public_id返回所有注册用户的列表和用户对象。
line 40 to 42 commits the changes to database.
line 40到42 line 40更改提交到数据库。

No need to use jsonify for formatting an object to JSON, Flask-restplus does it automatically

无需使用jsonify将对象格式化为JSON，Flask-restplus会自动执行

In the main package, create a new package called util . This package will contain all the necessary utilities we might need in our application.

在main软件包中，创建一个名为util的新软件包。该软件包将包含我们在应用程序中可能需要的所有必要实用程序。

In the util package, create a new file dto.py. As the name implies, the data transfer object (DTO) will be responsible for carrying data between processes. In our own case, it will be used for marshaling data for our API calls. We will understand this better as we proceed.

在util包中，创建一个新文件dto.py 顾名思义，数据传输对象( DTO )将负责在进程之间传送数据。在我们自己的情况下，它将用于封送我们API调用的数据。在进行过程中，我们将更好地理解这一点。

from flask_restplus import Namespace, fields


class UserDto:
    api = Namespace('user', description='user related operations')
    user = api.model('user', {
        'email': fields.String(required=True, description='user email address'),
        'username': fields.String(required=True, description='user username'),
        'password': fields.String(required=True, description='user password'),
        'public_id': fields.String(description='user Identifier')
    })

The above code within dto.py does the following:

dto.py中的上述代码执行以下操作：

line 5 creates a new namespace for user related operations. Flask-RESTPlus provides a way to use almost the same pattern as Blueprint. The main idea is to split your app into reusable namespaces. A namespace module will contain models and resources declaration.
line 5与用户相关的操作创建了一个新的命名空间。 Flask-RESTPlus提供了一种使用与Blueprint几乎相同的模式的方法。主要思想是将您的应用拆分为可重用的名称空间。命名空间模块将包含模型和资源声明。
line 6 creates a new user dto through the model interface provided by the api namespace in line 5.
line 6通过第line 5 api命名空间提供的model接口创建新用户dto。

User Controller: The user controller class handles all the incoming HTTP requests relating to the user .

用户控制器：用户控制器类处理与用户有关的所有传入HTTP请求。

Under the controller package, create a new file called user_controller.py with the following content:

在controller程序包下，创建一个名为user_controller.py的新文件，其内容如下：

from flask import request
from flask_restplus import Resource

from ..util.dto import UserDto
from ..service.user_service import save_new_user, get_all_users, get_a_user

api = UserDto.api
_user = UserDto.user


@api.route('/')
class UserList(Resource):
    @api.doc('list_of_registered_users')
    @api.marshal_list_with(_user, envelope='data')
    def get(self):
        """List all registered users"""
        return get_all_users()

    @api.response(201, 'User successfully created.')
    @api.doc('create a new user')
    @api.expect(_user, validate=True)
    def post(self):
        """Creates a new User """
        data = request.json
        return save_new_user(data=data)


@api.route('/<public_id>')
@api.param('public_id', 'The User identifier')
@api.response(404, 'User not found.')
class User(Resource):
    @api.doc('get a user')
    @api.marshal_with(_user)
    def get(self, public_id):
        """get a user given its identifier"""
        user = get_a_user(public_id)
        if not user:
            api.abort(404)
        else:
            return user

line 1 through 8 imports all the required resources for the user controller.We defined two concrete classes in our user controller which are userList and user. These two classes extends the abstract flask-restplus resource.

line 1至8导入了用户控制器所需的所有资源。我们在用户控制器中定义了两个具体的类userList和user 。这两个类扩展了抽象的flask-restplus资源。

Concrete resources should extend from this class and expose methods for each supported HTTP method. If a resource is invoked with an unsupported HTTP method, the API will return a response with status 405 Method Not Allowed. Otherwise the appropriate method is called and passed all arguments from the URL rule used when adding the resource to an API instance.

具体资源应从此类扩展， 并为每个受支持的HTTP方法公开方法。 如果使用不受支持的HTTP方法调用资源， 则API将返回状态为405方法不允许的响应。 否则，将调用适当的方法，并传递 将资源添加到API实例时使用的URL规则中的 所有参数 。

The api namespace in line 7 above provides the controller with several decorators which includes but is not limited to the following:

上面line 7的api名称空间为控制器提供了几个装饰器，包括但不限于以下几种：

api.route: A decorator to route resources
api。 route ： 用于路由资源的装饰器
api.marshal_with: A decorator specifying the fields to use for serialization (This is where we use the userDto we created earlier)
api。 marshal_with ： 一个装饰器，指定用于序列化的字段(这是我们使用 userDto 创建的 userDto )
api.marshal_list_with: A shortcut decorator for marshal_with above withas_list = True
api。 marshal_list_with： 一个快捷装饰为 marshal_with 与上述 as_list = True
api.doc: A decorator to add some api documentation to the decorated object
api。 doc ： 修饰器，用于向修饰对象添加一些api文档
api.response: A decorator to specify one of the expected responses
api。 响应： 装饰者指定预期的响应之一
api.expect: A decorator to Specify the expected input model ( we still use the userDto for the expected input)
api。 期望： 一个装饰器，用于指定期望的输入模型(我们仍将 userDto 用于期望的输入)
api.param: A decorator to specify one of the expected parameters
api。 param： 装饰器，用于指定预期参数之一

We have now defined our namespace with the user controller. Now its time to add it to the application entry point.

现在，我们已经使用用户控制器定义了名称空间。现在是时候将其添加到应用程序入口点了。

In the __init__.py file of app package, enter the following:

在app包的__init__.py文件中，输入以下内容：

# app/__init__.py

from flask_restplus import Api
from flask import Blueprint

from .main.controller.user_controller import api as user_ns

blueprint = Blueprint('api', __name__)

api = Api(blueprint,
          title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
          version='1.0',
          description='a boilerplate for flask restplus web service'
          )

api.add_namespace(user_ns, path='/user')

The above code within blueprint.py does the following:

blueprint.py中的上述代码执行以下操作：

In line 8, we create a blueprint instance by passing name and import_name. API is the main entry point for the application resources and hence needs to be initialized with the blueprint in line 10.
在line 8 ，我们通过传递name和import_name.创建一个蓝图实例import_name. API是应用程序资源的主要入口点，因此需要使用第line 10的blueprint进行初始化。
In line 16 , we add the user namespace user_ns to the list of namespaces in the API instance.
在line 16 ，我们将用户名称空间user_ns添加到API实例中的名称空间列表中。

We have now defined our blueprint. It’s time to register it on our Flask app.Update manage.py by importing blueprint and registering it with the Flask application instance.

现在，我们已经定义了蓝图。现在是时候在我们的Flask应用程序中注册它了。通过导入blueprint并将其注册到Flask应用程序实例中来更新manage.py 。

from app import blueprint
...

app = create_app(os.getenv('BOILERPLATE_ENV') or 'dev')
app.register_blueprint(blueprint)

app.app_context().push()

...

We can now test our application to see that everything is working fine.

现在我们可以测试我们的应用程序，看一切正常。

python manage.py run

Now open the URL http://127.0.0.1:5000 in your browser. You should see the swagger documentation.

现在，在浏览器中打开URL http://127.0.0.1:5000 。您应该看到招摇的文档。

Let’s test the create new user endpoint using the swagger testing functionality.

让我们使用招摇测试功能测试创建新用户端点。

You should get the following response

您应该得到以下回应

安全与认证 (Security and Authentication)

Let’s create a model blacklistToken for storing blacklisted tokens. In the models package, create a blacklist.py file with the following content:

让我们创建一个模型blacklistToken来存储列入黑名单的令牌。在models包中，创建一个具有以下内容的blacklist.py文件：

from .. import db
import datetime


class BlacklistToken(db.Model):
    """
    Token Model for storing JWT tokens
    """
    __tablename__ = 'blacklist_tokens'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    token = db.Column(db.String(500), unique=True, nullable=False)
    blacklisted_on = db.Column(db.DateTime, nullable=False)

    def __init__(self, token):
        self.token = token
        self.blacklisted_on = datetime.datetime.now()

    def __repr__(self):
        return '<id: token: {}'.format(self.token)

    @staticmethod
    def check_blacklist(auth_token):
        # check whether auth token has been blacklisted
        res = BlacklistToken.query.filter_by(token=str(auth_token)).first()
        if res:
            return True
        else:
            return False

Lets not forget to migrate the changes to take effect on our database. Import the blacklist class in manage.py.

别忘了迁移更改以对我们的数据库生效。将blacklist类导入manage.py 。

from app.main.model import blacklist

Run the migrate and upgrade commands

运行migrate和upgrade命令

python manage.py db migrate --message 'add blacklist table'

python manage.py db upgrade

Next create blacklist_service.py in the service package with the following content for blacklisting a token:

接下来，在服务包中创建blacklist_service.py ，其中包含以下内容以将令牌列入黑名单：

from app.main import db
from app.main.model.blacklist import BlacklistToken


def save_token(token):
    blacklist_token = BlacklistToken(token=token)
    try:
        # insert the token
        db.session.add(blacklist_token)
        db.session.commit()
        response_object = {
            'status': 'success',
            'message': 'Successfully logged out.'
        }
        return response_object, 200
    except Exception as e:
        response_object = {
            'status': 'fail',
            'message': e
        }
        return response_object, 200

Update the user model with two static methods for encoding and decoding tokens. Add the following imports:

使用两种用于编码和解码令牌的静态方法更新user模型。添加以下导入：

import datetime
import jwt
from app.main.model.blacklist import BlacklistToken
from ..config import key

Encoding
编码方式

def encode_auth_token(self, user_id):
        """
        Generates the Auth Token
        :return: string
        """
        try:
            payload = {
                'exp': datetime.datetime.utcnow() + datetime.timedelta(days=1, seconds=5),
                'iat': datetime.datetime.utcnow(),
                'sub': user_id
            }
            return jwt.encode(
                payload,
                key,
                algorithm='HS256'
            )
        except Exception as e:
            return e

Decoding: Blacklisted token, expired token and invalid token are taken into consideration while decoding the authentication token.
解码：在对身份验证令牌进行解码时，会考虑列入黑名单的令牌，过期的令牌和无效的令牌。

@staticmethod  
  def decode_auth_token(auth_token):
        """
        Decodes the auth token
        :param auth_token:
        :return: integer|string
        """
        try:
            payload = jwt.decode(auth_token, key)
            is_blacklisted_token = BlacklistToken.check_blacklist(auth_token)
            if is_blacklisted_token:
                return 'Token blacklisted. Please log in again.'
            else:
                return payload['sub']
        except jwt.ExpiredSignatureError:
            return 'Signature expired. Please log in again.'
        except jwt.InvalidTokenError:
            return 'Invalid token. Please log in again.'

Now let’s write a test for the user model to ensure that our encode and decode functions are working properly.

现在让我们为user模型编写一个测试，以确保我们的encode和decode功能正常运行。

In the test package, create base.py file with the following content:

在test包中，创建具有以下内容的base.py文件：

from flask_testing import TestCase
from app.main import db
from manage import app


class BaseTestCase(TestCase):
    """ Base Tests """

    def create_app(self):
        app.config.from_object('app.main.config.TestingConfig')
        return app

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

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

The BaseTestCase sets up our test environment ready before and after every test case that extends it.

BaseTestCase在扩展每个测试用例之前和之后设置我们的测试环境。

Create test_user_medol.py with the following test cases:

使用以下测试用例创建test_user_medol.py ：

import unittest
import datetime

from app.main import db
from app.main.model.user import User
from app.test.base import BaseTestCase


class TestUserModel(BaseTestCase):

    def test_encode_auth_token(self):
        user = User(
            email='test@test.com',
            password='test',
            registered_on=datetime.datetime.utcnow()
        )
        db.session.add(user)
        db.session.commit()
        auth_token = user.encode_auth_token(user.id)
        self.assertTrue(isinstance(auth_token, bytes))

    def test_decode_auth_token(self):
        user = User(
            email='test@test.com',
            password='test',
            registered_on=datetime.datetime.utcnow()
        )
        db.session.add(user)
        db.session.commit()
        auth_token = user.encode_auth_token(user.id)
        self.assertTrue(isinstance(auth_token, bytes))
        self.assertTrue(User.decode_auth_token(auth_token.decode("utf-8") ) == 1)


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

Run the test with python manage.py test. All the tests should pass.

使用python manage.py test运行python manage.py test 。所有测试都应该通过。

Let’s create the authentication endpoints for login and logout.

让我们为登录和注销创建身份验证端点 。

First we need a dto for the login payload. We will use the auth dto for the @expect annotation in login endpoint. Add the code below to the dto.py
首先，我们需要一个dto作为登录有效负载。我们将在login端点中的@expect注释中使用auth dto。将以下代码添加到dto.py

class AuthDto:
    api = Namespace('auth', description='authentication related operations')
    user_auth = api.model('auth_details', {
        'email': fields.String(required=True, description='The email address'),
        'password': fields.String(required=True, description='The user password '),
    })

Next, we create an authentication helper class for handling all authentication related operations. This auth_helper.py will be in the service package and will contain two static methods which are login_user and logout_user
接下来，我们创建一个身份验证帮助程序类，以处理所有与身份验证相关的操作。该auth_helper.py将位于服务包中，并将包含两个静态方法，分别是login_user和logout_user

When a user is logged out, the user’s token is blacklisted ie the user can’t log in again with that same token.

当用户注销时，该用户的令牌将被列入黑名单，即该用户无法使用相同的令牌再次登录。

from app.main.model.user import User
from ..service.blacklist_service import save_token


class Auth:

    @staticmethod
    def login_user(data):
        try:
            # fetch the user data
            user = User.query.filter_by(email=data.get('email')).first()
            if user and user.check_password(data.get('password')):
                auth_token = user.encode_auth_token(user.id)
                if auth_token:
                    response_object = {
                        'status': 'success',
                        'message': 'Successfully logged in.',
                        'Authorization': auth_token.decode()
                    }
                    return response_object, 200
            else:
                response_object = {
                    'status': 'fail',
                    'message': 'email or password does not match.'
                }
                return response_object, 401

        except Exception as e:
            print(e)
            response_object = {
                'status': 'fail',
                'message': 'Try again'
            }
            return response_object, 500

    @staticmethod
    def logout_user(data):
        if data:
            auth_token = data.split(" ")[1]
        else:
            auth_token = ''
        if auth_token:
            resp = User.decode_auth_token(auth_token)
            if not isinstance(resp, str):
                # mark the token as blacklisted
                return save_token(token=auth_token)
            else:
                response_object = {
                    'status': 'fail',
                    'message': resp
                }
                return response_object, 401
        else:
            response_object = {
                'status': 'fail',
                'message': 'Provide a valid auth token.'
            }
            return response_object, 403

Let us now create endpoints for login and logout operations.
现在让我们为login和logout操作创建端点。

In the controller package, create
在控制器包中，创建

auth_controller.py with the following contents:
auth_controller.py具有以下内容：

from flask import request
from flask_restplus import Resource

from app.main.service.auth_helper import Auth
from ..util.dto import AuthDto

api = AuthDto.api
user_auth = AuthDto.user_auth


@api.route('/login')
class UserLogin(Resource):
    """
        User Login Resource
    """
    @api.doc('user login')
    @api.expect(user_auth, validate=True)
    def post(self):
        # get the post data
        post_data = request.json
        return Auth.login_user(data=post_data)


@api.route('/logout')
class LogoutAPI(Resource):
    """
    Logout Resource
    """
    @api.doc('logout a user')
    def post(self):
        # get auth token
        auth_header = request.headers.get('Authorization')
        return Auth.logout_user(data=auth_header)

At this point the only thing left is to register the auth api namespace with the application Blueprint
此时，剩下的唯一事情就是向应用程序Blueprint注册auth api名称空间

Update __init__.py file of app package with the following

使用以下更新app包的__init__.py文件

# app/__init__.py

from flask_restplus import Api
from flask import Blueprint

from .main.controller.user_controller import api as user_ns
from .main.controller.auth_controller import api as auth_ns

blueprint = Blueprint('api', __name__)

api = Api(blueprint,
          title='FLASK RESTPLUS API BOILER-PLATE WITH JWT',
          version='1.0',
          description='a boilerplate for flask restplus web service'
          )

api.add_namespace(user_ns, path='/user')
api.add_namespace(auth_ns)

Run the application with python manage.py run and open the url http://127.0.0.1:5000 in your browser.

使用python manage.py run运行应用程序，然后在浏览器中打开URL http://127.0.0.1:5000 。

The swagger documentation should now reflect the newly created auth namespace with the login and logout endpoints.

昂首阔步的文档现在应该反映具有login和logout端点的新创建的auth名称空间。

Before we write some tests to ensure our authentication is working as expected, let’s modify our registration endpoint to automatically login a user once the registration is successful.

在编写一些测试以确保我们的身份验证按预期工作之前，让我们修改注册端点，以在注册成功后自动登录用户。

Add the method generate_token below to user_service.py:

将下面的方法generate_token添加到user_service.py ：

def generate_token(user):
    try:
        # generate the auth token
        auth_token = user.encode_auth_token(user.id)
        response_object = {
            'status': 'success',
            'message': 'Successfully registered.',
            'Authorization': auth_token.decode()
        }
        return response_object, 201
    except Exception as e:
        response_object = {
            'status': 'fail',
            'message': 'Some error occurred. Please try again.'
        }
        return response_object, 401

The generate_token method generates an authentication token by encoding the user id. This token is the returned as a response.

generate_token方法通过对用户id.进行编码来生成身份验证令牌 id. 该令牌是作为响应返回。

Next, replace the return block in save_new_user method below

接下来，在下面的save_new_user方法中替换return块

response_object = {
    'status': 'success',
    'message': 'Successfully registered.'
}
return response_object, 201

with

与

return generate_token(new_user)

Now its time to test the login and logout functionalities. Create a new test file test_auth.py in the test package with the following content:

现在该测试login和logout功能了。在测试包中使用以下内容创建一个新的测试文件test_auth.py ：

import unittest
import json
from app.test.base import BaseTestCase


def register_user(self):
    return self.client.post(
        '/user/',
        data=json.dumps(dict(
            email='example@gmail.com',
            username='username',
            password='123456'
        )),
        content_type='application/json'
    )


def login_user(self):
    return self.client.post(
        '/auth/login',
        data=json.dumps(dict(
            email='example@gmail.com',
            password='123456'
        )),
        content_type='application/json'
    )


class TestAuthBlueprint(BaseTestCase):

    def test_registered_user_login(self):
            """ Test for login of registered-user login """
            with self.client:
                # user registration
                user_response = register_user(self)
                response_data = json.loads(user_response.data.decode())
                self.assertTrue(response_data['Authorization'])
                self.assertEqual(user_response.status_code, 201)

                # registered user login
                login_response = login_user(self)
                data = json.loads(login_response.data.decode())
                self.assertTrue(data['Authorization'])
                self.assertEqual(login_response.status_code, 200)

    def test_valid_logout(self):
        """ Test for logout before token expires """
        with self.client:
            # user registration
            user_response = register_user(self)
            response_data = json.loads(user_response.data.decode())
            self.assertTrue(response_data['Authorization'])
            self.assertEqual(user_response.status_code, 201)

            # registered user login
            login_response = login_user(self)
            data = json.loads(login_response.data.decode())
            self.assertTrue(data['Authorization'])
            self.assertEqual(login_response.status_code, 200)

            # valid token logout
            response = self.client.post(
                '/auth/logout',
                headers=dict(
                    Authorization='Bearer ' + json.loads(
                        login_response.data.decode()
                    )['Authorization']
                )
            )
            data = json.loads(response.data.decode())
            self.assertTrue(data['status'] == 'success')
            self.assertEqual(response.status_code, 200)

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

Visit the github repo for a more exhaustive test cases.

访问github回购以获得更详尽的测试案例。

路线保护和授权 (Route protection and Authorization)

So far, we have successfully created our endpoints, implemented login and logout functionalities but our endpoints remains unprotected.

到目前为止，我们已经成功创建了端点，实现了登录和注销功能，但是端点仍然不受保护。

We need a way to define rules that determines which of our endpoint is open or requires authentication or even an admin privilege.

我们需要一种方法来定义规则，这些规则确定哪个端点是打开的或需要身份验证甚至是管理员特权。

We can achieve this by creating custom decorators for our endpoints.

我们可以通过为端点创建自定义装饰器来实现。

Before we can protect or authorize any of our endpoints, we need to know the currently logged in user. We can do this by pulling the Authorization token from the header of the current request by using the flask library request.We then decode the user details from the Authorization token.

在我们可以保护或授权任何端点之前，我们需要知道当前登录的用户。我们可以通过使用flask库request.从当前请求的标头中提取Authorization token来实现此目的request. 然后，我们从Authorization token解码用户详细信息。

In the Auth class of auth_helper.py file, add the following static method:

在auth_helper.py文件的Auth类中，添加以下静态方法：

@staticmethod
def get_logged_in_user(new_request):
        # get the auth token
        auth_token = new_request.headers.get('Authorization')
        if auth_token:
            resp = User.decode_auth_token(auth_token)
            if not isinstance(resp, str):
                user = User.query.filter_by(id=resp).first()
                response_object = {
                    'status': 'success',
                    'data': {
                        'user_id': user.id,
                        'email': user.email,
                        'admin': user.admin,
                        'registered_on': str(user.registered_on)
                    }
                }
                return response_object, 200
            response_object = {
                'status': 'fail',
                'message': resp
            }
            return response_object, 401
        else:
            response_object = {
                'status': 'fail',
                'message': 'Provide a valid auth token.'
            }
            return response_object, 401

Now that we can retrieve the logged in user from the request, let’s go ahead and create the decorators.

现在，我们可以从请求中检索登录的用户，让我们继续创建decorators.

Create a file decorator.py in the util package with the following content:

使用以下内容在util包中创建一个文件decorator.py ：

from functools import wraps
from flask import request

from app.main.service.auth_helper import Auth


def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):

        data, status = Auth.get_logged_in_user(request)
        token = data.get('data')

        if not token:
            return data, status

        return f(*args, **kwargs)

    return decorated


def admin_token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):

        data, status = Auth.get_logged_in_user(request)
        token = data.get('data')

        if not token:
            return data, status

        admin = token.get('admin')
        if not admin:
            response_object = {
                'status': 'fail',
                'message': 'admin token required'
            }
            return response_object, 401

        return f(*args, **kwargs)

    return decorated

For more information about decorators and how to create them, take a look at this link.

有关装饰器及其创建方法的更多信息，请查看此链接。

Now that we have created the decorators token_required and admin_token_required for valid token and for an admin token respectively, all that is left is to annotate the endpoints which we wish to protect with the freecodecamp orgappropriate decorator.

现在，我们已经admin_token_required为有效令牌和管理员令牌创建了装饰器token_required和admin_token_required ，剩下的就是用freecodecamp orgappropriate 装饰器注释我们希望保护的端点。

额外提示 (Extra tips)

Currently to perform some tasks in our application, we are required to run different commands for starting the app, running tests, installing dependencies etc. We can automate those processes by arranging all the commands in one file using Makefile.

当前，要在我们的应用程序中执行某些任务，我们需要运行不同的命令来启动应用程序，运行测试，安装依赖项等。我们可以通过使用Makefile.将所有命令排列在一个文件中来自动化这些过程Makefile.

On the root directory of the application, create a Makefile with no file extension. The file should contain the following:

在应用程序的根目录上，创建一个没有文件扩展名的Makefile 。该文件应包含以下内容：

.PHONY: clean system-packages python-packages install tests run all

clean:
   find . -type f -name '*.pyc' -delete
   find . -type f -name '*.log' -delete

system-packages:
   sudo apt install python-pip -y

python-packages:
   pip install -r requirements.txt

install: system-packages python-packages

tests:
   python manage.py test

run:
   python manage.py run

all: clean install tests run

Here are the options of the make file.

这是make文件的选项。

make install : installs both system-packages and python-packages
make install ：同时安装系统软件包和python软件包
make clean : cleans up the app
make clean ：清理应用程序
make tests : runs the all the tests
make tests ：运行所有测试
make run : starts the application
make run ：启动应用程序
make all : performs clean-up,installation , run tests , and starts the app.
make all ：执行clean-up ， installation ，运行tests ，然后starts应用程序。

扩展应用程序和结论 (Extending the App & Conclusion)

It’s pretty easy to copy the current application structure and extend it to add more functionalities/endpoints to the App. Just view any of the previous routes that have been implemented.

复制当前应用程序结构并对其进行扩展以为该应用程序添加更多功能/端点非常容易。只需查看之前已实施的任何路由即可。

Feel free to leave a comment have you any question, observations or recommendations. Also, if this post was helpful to you, click on the clap icon so others will see this here and benefit as well.

如有任何问题，意见或建议，请随时发表评论。 另外，如果该帖子对您有所帮助，请单击拍手图标，这样其他人也会在这里看到并受益。

Visit the github repository for the complete project.

访问github存储库以获取完整的项目。

Thanks for reading and good luck!

感谢您的阅读和好运！

翻译自: https://www.freecodecamp.org/news/structuring-a-flask-restplus-web-service-for-production-builds-c2ec676de563/