按照上一章中的示例，我们将设计为单体的系统拆分为更小的服务。本章我们将详细分析上一章提到的其中一个微服务（Thoughts Backend）。

We 将讨论如何使用 Python 将这个微服务开发为应用程序。该微服务将准备好通过标准的 Web RESTful 接口与其他微服务进行交互，使其成为我们全球微服务架构系统的基础。

我们将讨论不同的元素，例如 API 设计、支持它的数据库架构，以及如何实现以及如何实现微服务。最后，我们将了解如何测试应用程序以确保其正常工作。

本章将涵盖以下主题：

• Analyzing the Thoughts Backend microservice
• Designing the RESTful API
• Defining the database schema
• Implementing the service
• Testing the code

在本章结束时，您将了解如何成功开发微服务应用程序，包括从设计到测试的不同阶段。

Thoughts 后端示例可以在此处找到 （https://github.com/PacktPublishing/Hands-On-Docker-for-Microservices-with-Python/tree/master/Chapter02/ThoughtsBackend)<跨度>。安装和运行说明可以在其 README.md 文件中找到。

让我们记住我们在上一章中创建的微服务图：

该图显示了我们示例系统的不同元素：两个后端，用户和想法，以及 HTML 前端。

Thoughts Backend 将负责存储新想法、检索现有想法和搜索数据库。

由于 Thoughts Backend 将在外部可用，我们需要实现一个安全层。这意味着我们需要识别产生操作的用户并验证其有效性。对于这个服务示例，我们将从登录用户创建一个新想法，我们将检索我的想法，由当前登录用户创建的想法。

这个安全层将以标题的形式出现。此标头将包含由用户后端签名的信息，以验证其来源。它将采用 JSON Web Token (JWT)、https://jwt.io/introduction/，这是用于此目的的标准。

这个方法d应该由t他的Users Backend团队处理，并打包以便其他微服务可以使用它。在本章中，我们将在此微服务中包含代码，但稍后我们将看到如何创建它以使其与用户后端相关。

如果请求没有有效的标头，API 将返回 401 Unauthorized 状态码。

了解了身份验证系统的工作原理后，我们就可以开始设计 API 接口了。

我们的 API 将遵循 RESTful 设计原则。这意味着我们将使用构建的 URI 来表示资源，然后使用 HTTP 方法对这些资源执行操作。

RESTful API 的主要属性之一是请求需要是无状态的，这意味着每个请求都是完全独立的，可以由任何服务器提供服务。所有必需的数据都应该在客户端（将其附加到请求中发送）或在数据库中（因此服务器将完整检索它）。在处理 Docker 容器时，此属性是一项硬性要求，因为它们可以在没有警告的情况下被销毁和重新创建。

在开始 API 设计之前有一个关于 REST 的简短提醒总是好的，所以您可以查看 https://restfulapi.net/ 回顾一下。

我们的API接口如下：

注意 API 有两个元素：

• A public API, starting with /api:
  • An authenticated public API, starting with /api/me. The user needs to be authenticated to perform these actions. A non-authenticated request will return a 401 Unauthorized status code.
  • A non-authenticated public API, starting with /api. Any user, even not authenticated, can perform these actions.
• An admin API (starting with /admin). This won't be exposed publicly. It spares the authentication and allows you to do operations that are not designed to be done by customers. Clearly labeling with a prefix helps to audit the operations and clearly signifies that they should not be available outside of your data center.

一个想法的格式如下：

thought
{
    id integer
    username string
    text string
    timestamp string($date-time)
}

要创建一个，只需要发送文本。自动设置时间戳，自动创建 ID，并通过身份验证数据检测用户名。

最后一个细节：关于是否最好用最后的斜杠结束 URI 资源存在一些争论。但是，在使用 Flask 时，使用斜杠定义它们将为没有正确结尾的请求返回重定向状态代码 308 PERMANENT_REDIRECT。在任何情况下，尽量保持一致以避免混淆。

数据库模式很简单，并且继承自单体应用程序。我们只关心想法，存储在thought_model表中，所以数据库结构如下：

该表以 thoughts_backend/models.py 文件中的代码表示，以 SQLAlchemy 格式描述，代码如下：

class ThoughtModel(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50))
    text = db.Column(db.String(250))
    timestamp = db.Column(db.DateTime, server_default=func.now())

SQLAlchemy 能够为测试目的或开发模式创建表。对于本章，我们将数据库定义为 SQLite，它将数据存储在 db.sqlite3 文件中。

SQLAlchemy (https://www.sqlalchemy.org/) 是一个强大的 Python 模块，用于处理 SQL 数据库。使用 Python 等高级语言处理数据库有两种方法。一种是保留低级方法并执行原始 SQL 语句，检索数据库中的数据。另一种是使用 Object-Relational Mapper (ORM) 抽象数据库并使用接口，而无需深入了解其实现细节。

Python 数据库 API 规范 (PEP 249—https ://www.python.org/dev/peps/pep-0249/)，后面是所有主要数据库，如psycopg2(http://initd.org/psycopg/) 用于 PostgreSQL。这主要是创建 SQL 字符串命令，执行它们，然后解析结果。这使我们能够定制每个查询，但对于一遍又一遍地重复的常见操作来说，它的效率并不高。 PonyORM (https://ponyorm.org/) 是另一个级别不低但仍旨在复制 SQL 的示例语法和结构。

对于第二种方法，最著名的例子可能是 Django ORM (https ://docs.djangoproject.com/en/2.2/topics/db/）。它使用定义的模型 python 对象抽象数据库访问。它对于常见操作非常有效，但它的模型假设数据库的定义是在我们的 Python 代码中完成的，映射遗留数据库可能非常痛苦。由 ORM 创建的一些复杂的 SQL 操作可能会花费大量时间，而定制的查询可以节省大量时间。在没有意识到的情况下执行慢速查询也很容易，因为该工具将我们从最终结果中抽象出来。

SQLAlchemy (https://www.sqlalchemy.org/) 非常灵活，可以在两端工作光谱。它不像 Django ORM 那样简单易用，但它允许我们将现有数据库映射到 ORM 中。这就是我们将在示例中使用它的原因：它可以采用现有的、复杂的遗留数据库并对其进行映射，从而使您可以轻松地执行简单的操作，并以您想要的方式执行复杂的操作。

Flask-SQLAlchemy 的文档（https://flask-sqlalchemy.palletsprojects.com/en /2.x/) 是一个很好的起点，因为它总结了主要操作，而且主要的 SQLAlchemy 文档一开始可能会让人不知所措。

定义模型后，我们可以使用模型中的 query 属性执行查询并进行相应的过滤：

# Retrieve a single thought by its primary key
thought = ThoughtModel.query.get(thought_id)
# Retrieve all thoughts filtered by a username
thoughts = ThoughtModel.query.filter_by(username=username)
.order_by('id').all()

存储和删除一行需要使用会话然后提交它：

# Create a new thought
new_thought = ThoughtModel(username=username, text=text, timestamp=datetime.utcnow())
db.session.add(new_thought)
db.session.commit()

# Retrieve and delete a thought
thought = ThoughtModel.query.get(thought_id)
db.session.delete(thought)
db.session.commit()

要查看如何配置数据库访问，请查看 thoughts_backend/db.py 文件。

为了实现这个微服务，我们将使用 Flask-RESTPlus (https://flask-restplus.readthedocs. io/en/stable/）。这是一个 Flask (https://palletsprojects.com/p/flask/) 扩展。 Flask 是一个著名的用于 Web 应用程序的 Python 微框架，特别擅长实现微服务，因为它体积小、易于使用，并且与 Web 应用程序方面的常用技术堆栈兼容，因为它使用 Web 服务器网关接口 (WSGI) 协议。

Flask 能够实现 RESTful 接口，但 Flask-RESTPlus 添加了一些非常有趣的功能，可以实现良好的开发实践和开发速度：

• It defines namespaces, which are ways of creating prefixes and structuring the code. This helps long-term maintenance and helps with the design when creating new endpoints.

• It has a full solution for parsing input parameters. This means that we have an easy way of dealing with endpoints that requires several parameters and validates them. Using the Request Parsing (https://flask-restplus.readthedocs.io/en/stable/parsing.html) module is similar to using the argparse command-line module (https://docs.python.org/3/library/argparse.html) that's included in the Python standard library. It allows defining arguments in the body of the request, headers, query strings, or even cookies.
• In the same way, it has a serialization framework for the resulting objects. Flask-RESTful calls it response marshalling (https://flask-restplus.readthedocs.io/en/stable/marshalling.html). This helps to define objects that can be reused, clarifying the interface and simplifying the development. If enabled, it also allows for field masks, which return partial objects.
• It has full Swagger API documentation support. Swagger (https://swagger.io/) is an open source project to help in the design, implementation, documentation, and testing of RESTful API web services, following standard OpenAPI specifications. Flask-RESTPlus automatically generates a Swagger specification and self-documenting page:

Flask 的其他不错的元素源于它是一个受欢迎的项目并且有很多支持的工具：

• We will use the connector for SQLAlchemy, Flask-SQLAlchemy (https://flask-sqlalchemy.palletsprojects.com/en/2.x/). Its documentation covers most of the common cases, while the SQLAlchemy documentation is more detailed and can be a bit overwhelming.
• To run the tests, the pytest-flask module (https://pytest-flask.readthedocs.io/en/latest/) creates some fixtures ready to work with a Flask application. We will talk more about this in the Testing the code section.

典型的 RESTful 应用程序具有以下一般结构：

1. A URL-defined resource. This resource allows one or more actions through HTTP methods (GET, POST, and so on).
2. When each of the actions is called, the framework routes the request until the defined code executes the action.
3. If there are any input parameters, they'll need to be validated first.
4. Perform the action and obtain a result value. This action will normally involve one or more calls to the database, which will be done in the shape of models.
5. Prepare the resulting result value and encode it in a way that's understood by the client, typically in JSON.
6. Return the encoded value to the client with the adequate status code.

大多数这些操作都是由框架完成的。需要完成一些配置工作，但这是我们的 Web 框架（本示例中的 Flask-RESTPlus）最能提供帮助的地方。特别是，除了 第 4 步 之外的所有内容都将大大简化。

让我们看一个简单的代码示例（可在 GitHub 中找到）来描述它：

api_namespace = Namespace('api', description='API operations')

@api_namespace.route('/thoughts/<int:thought_id>/')
class ThoughtsRetrieve(Resource):

    @api_namespace.doc('retrieve_thought')
    @api_namespace.marshal_with(thought_model)
    def get(self, thought_id):
        '''
        Retrieve a thought
        '''
        thought = ThoughtModel.query.get(thought_id)
        if not thought:
            # The thought is not present
            return '', http.client.NOT_FOUND

        return thought

这实现了 GET /api/thoughts/X/ 操作，通过 ID 检索单个想法。

让我们分析每个元素。请注意，这些行按主题分组：

1. First, we define the resource by its URL. Note that api_namespace sets the api prefix to the URL, which validates that parameter X is an integer:

api_namespace = Namespace('api', description='API operations')

@api_namespace.route('/thoughts/<int:thought_id>/')
class ThoughtsRetrieve(Resource):
    ...

2. The class allows you to perform multiple actions on the same resource. In this case, we only do one: the GET action.
3. Note that the thought_id parameter, encoded in the URL, is passed as a parameter to the method:

class ThoughtsRetrieve(Resource):

    def get(self, thought_id):
        ...

4. We can now execute the action, which is a search in the database to retrieve a single object. Call ThoughModel to search for the specified thought. If found, it's returned with a http.client.OK (200) status code. If it's not found, an empty result and a http.client.NOT_FOUND 404 status code is returned:

def get(self, thought_id):
    thought = ThoughtModel.query.get(thought_id)
    if not thought:
        # The thought is not present
        return '', http.client.NOT_FOUND

    return thought

5. The thought object is being returned. The marshal_with decorator describes how the Python object should be serialized into a JSON structure. We'll see later how to configure it:

@api_namespace.marshal_with(thought_model)
def get(self, thought_id):
    ...
    return thought

6. Finally, we have some documentation, including the docstring that will be rendered by the autogenerated Swagger API:

class ThoughtsRetrieve(Resource):

    @api_namespace.doc('retrieve_thought')
    def get(self, thought_id):
        '''
        Retrieve a thought
        '''
        ...

如您所见，大部分操作都是通过 Flask-RESTPlus 配置和执行的，作为开发人员的大部分工作是繁琐的步骤 4。但是，还有一些工作要做，即配置预期的输入参数并验证它们，以及如何将返回的对象序列化为正确的 JSON。我们将看到 Flask-RESTPlus 如何帮助我们解决这个问题。

输入参数可以采用不同的形状。说到输入参数，我们主要讲两种：

• String query parameters encoded into the URL. These are normally used for the GET requests, and look like the following:

http://test.com/some/path?param1=X&param2=Y

它们是 URL 的一部分，并将存储在沿途的任何日志中。参数被编码成自己的格式，称为 URL 编码 (https://www. urlencoder.io/learn/）。您可能已经注意到，例如，一个空白空间被转换为 %20。

• Let's look at the body of the HTTP request. This is typically used in the POST and PUT requests. The specific format can be specified using the Content-Type header. By default, the Content-Type header is defined as application/x-www-form-urlencoded, which encodes it in URL encoding. In modern applications, this is replaced with application/json to encode them in JSON.

但是还有另外两个地方也可以使用传递参数：

• As a part of the URL: Things such as thought id are parameters. Try to follow RESTful principles and define your URLs as resources to avoid confusion. Query parameters are best left as optional.
• Headers: Normally, headers give information about metadata, such as the format of the request, the expected format, or authentication data. But they need to be treated as input parameters as well.

所有这些元素都由 Flask-RESTPlus 自动解码，因此我们不需要处理编码和低级访问。

让我们看看它在我们的示例中是如何工作的。这段代码是从 GitHub 中的代码中提取出来的，并被缩短以描述 解析 参数：

authentication_parser = api_namespace.parser()
authentication_parser.add_argument('Authorization', 
location='headers', type=str, help='Bearer Access 
Token')

thought_parser = authentication_parser.copy()
thought_parser.add_argument('text', type=str, required=True, help='Text of the thought')

@api_namespace.route('/me/thoughts/')
class MeThoughtListCreate(Resource):

    @api_namespace.expect(thought_parser)
    def post(self):
        args = thought_parser.parse_args()
        username = authentication_header_parser(args['Authorization'])
        text=args['text']
        ...

我们在以下几行中定义了一个解析器：

authentication_parser = api_namespace.parser()
authentication_parser.add_argument('Authorization', 
location='headers', type=str, help='Bearer Access Token')

thought_parser = authentication_parser.copy()
thought_parser.add_argument('text', type=str, required=True, help='Text of the thought')

authentication_parser 由 thought_parser 继承，以扩展功能并将两者结合。每个参数都根据类型以及是否需要进行定义。如果缺少必需的参数或其他元素不正确，Flask-RESTPlus 将引发 400 BAD_REQUEST 错误，并提供有关问题的反馈。

因为我们想以稍微不同的方式处理身份验证，我们将其标记为不需要并允许它使用默认值（为框架创建的）None。请注意，我们指定 Authorization 参数应位于标头中。

post 方法获取一个装饰器以表明它需要 thought_parser 参数，我们使用 parse_args 对其进行解析：

@api_namespace.route('/me/thoughts/')
class MeThoughtListCreate(Resource):

    @api_namespace.expect(thought_parser)
    def post(self):
        args = thought_parser.parse_args()
        ...

此外，args 现在是一个字典，所有参数都已正确解析并在下一行中使用。

在身份验证标头的特定情况下，有一个特定的函数可以使用它，它通过 abort 的使用返回一个 401 UNAUTHORIZED 状态代码。此调用立即停止请求：

def authentication_header_parser(value):
    username = validate_token_header(value, config.PUBLIC_KEY)
    if username is None:
        abort(401)
    return username


class MeThoughtListCreate(Resource):

    @api_namespace.expect(thought_parser)
    def post(self):
       args = thought_parser.parse_args()
       username = authentication_header_parser(args['Authentication'])
       ...

我们将暂时搁置要执行的操作（在数据库中存储一个新想法），并专注于其他框架配置，将结果序列化为 JSON 对象。

我们需要返回我们的结果。最简单的方法是通过序列化程序或编组模型定义 JSON 结果应具有的形状（https://flask-restplus.readthedocs.io/en/stable/marshalling.html)。

序列化器模型被定义为具有预期字段和字段类型的字典：

from flask_restplus import fields

model = {
    'id': fields.Integer(),
    'username': fields.String(),
    'text': fields.String(),
    'timestamp': fields.DateTime(),
}
thought_model = api_namespace.model('Thought', model)

该模型将采用 Python 对象，并将每个属性转换为相应的 JSON 元素，如字段中所定义：

@api_namespace.route('/me/thoughts/')
class MeThoughtListCreate(Resource):

    @api_namespace.marshal_with(thought_model)
    def post(self):
        ...
        new_thought = ThoughtModel(...)
        return new_thought

请注意，new_thought 是一个 ThoughtModel 对象，由 SQLAlchemy 检索。我们接下来会详细了解它，但现在，只要说它具有模型中定义的所有属性就足够了：id、username、text 和 时间戳。

默认情况下，内存对象中不存在的任何属性都将具有 None 的值。您可以将此默认值更改为将返回的值。您可以指定一个函数，以便在生成响应时调用它来检索一个值。这是向对象添加动态信息的一种方式：

model = {
    'timestamp': fields.DateTime(default=datetime.utcnow),
}

您还可以添加要序列化的属性的名称，以防它与预期结果不同，或者添加将被调用以检索值的 lambda 函数：

model = {
    'thought_text': fields.String(attribute='text'),
    'thought_username': fields.String(attribute=lambda x: x.username),
 }

对于更复杂的对象，您可以像这样嵌套值。请注意，这从文档的角度定义了两个模型，并且每个 Nested 元素都会创建一个新范围。您还可以使用 List 添加多个相同类型的实例：

extra = {
   'info': fields.String(),
}
extra_info = api_namespace.model('ExtraInfo', extra)

model = {
    'extra': fields.Nested(extra),
    'extra_list': fields.List(fields.Nested(extra)),
 }

一些可用字段有更多选项，例如 DateTime 字段的日期格式。检查完整字段的文档（https://flask-restplus.readthedocs.io /en/stable/api.html#models）了解更多详情。

如果返回元素列表，请在 marshal_with 装饰器中添加 as_list=True 参数：

@api_namespace.route('/me/thoughts/')
class MeThoughtListCreate(Resource):

    @api_namespace.marshal_with(thought_model, as_list=True)
    def get(self):
        ...
        thoughts = (
            ThoughtModel.query.filter(
                ThoughtModel.username == username
            )
            .order_by('id').all()
        )
        return thoughts

marshal_with 装饰器会将 result 对象从 Python 对象转换为相应的 JSON 数据对象。

默认情况下，它会返回一个 http.client.OK (200) 状态码，但是我们可以返回一个不同的状态码，返回两个值：第一个是 marshal 第二个是状态码。 marshal_with 装饰器中的 code 参数用于文档目的。注意，在这种情况下，我们需要添加特定的 marshal 调用：

@api_namespace.route('/me/thoughts/')
class MeThoughtListCreate(Resource):

    @api_namespace.marshal_with(thought_model, 
         code=http.client.CREATED)
    def post(self):
        ...
        result = api_namespace.marshal(new_thought, thought_model)
        return result, http.client.CREATED

Swagger 文档将显示所有使用定义的 marshal 对象：

有关更多详细信息，您可以在 https://flask-restplus 查看完整的编组文档.readthedocs.io/en/stable/marshalling.html) 的 Flask-RESTPlus。

最后，我们到达输入数据干净且可以使用的特定部分，并且我们知道如何返回结果。这部分可能涉及执行一些数据库查询或查询并组合结果。让我们看下面的例子：

@api_namespace.route('/thoughts/')
class ThoughtList(Resource):

    @api_namespace.doc('list_thoughts')
    @api_namespace.marshal_with(thought_model, as_list=True)
    @api_namespace.expect(search_parser)
    def get(self):
        '''
        Retrieves all the thoughts
        '''
        args = search_parser.parse_args()
        search_param = args['search']
        # Action
        query = ThoughtModel.query
        if search_param:
            query =(query.filter(
                ThoughtModel.text.contains(search_param)))

        query = query.order_by('id')
        thoughts = query.all()
        # Return the result
        return thoughts

您可以在这里看到，在解析参数后，我们使用 SQLAlchemy 检索查询，如果存在 search 参数，将应用过滤器。我们使用 all() 获得所有结果，返回所有 ThoughtModel 对象。

正如我们在 marshal_with 装饰器中指定的那样，返回对象会自动编组（将它们编码为 JSON）。

身份验证的逻辑封装在 thoughts_backend/token_validation.py 文件中。这包含标头的生成和验证。

以下函数生成 Bearer 令牌：

def encode_token(payload, private_key):
    return jwt.encode(payload, private_key, algorithm='RS256')

def generate_token_header(username, private_key):
    '''
    Generate a token header base on the username. 
    Sign using the private key.
    '''
    payload = {
        'username': username,
        'iat': datetime.utcnow(),
        'exp': datetime.utcnow() + timedelta(days=2),
    }
    token = encode_token(payload, private_key)
    token = token.decode('utf8')
    return f'Bearer {token}'

这会生成一个 JWT 有效负载。它包括要用作自定义值的 username，但它还添加了两个标准字段，一个 exp expiration date 和 iat 令牌的生成时间。

然后使用 RS256 算法使用私钥对令牌进行编码，并以正确的格式返回：Bearer 。

反向操作是从编码的标头中获取用户名。这里的代码更长，因为我们应该考虑到我们可能会收到 Authentication 标头的不同选项。该标头直接来自我们的公共 API，因此我们应该期望任何值和程序都为它做好防御准备。

令牌本身的解码很简单，因为 jwt.decode 操作将执行此操作：

def decode_token(token, public_key):
    return jwt.decode(token, public_key, algoritms='RS256')

但是在到达这一步之前，我们需要获取token并验证header在多种情况下是否有效，所以我们首先检查header是否为空，以及它是否具有正确的格式，提取token：

def validate_token_header(header, public_key):
    if not header:
        logger.info('No header')
        return None

    # Retrieve the Bearer token
    parse_result = parse('Bearer {}', header)
    if not parse_result:
        logger.info(f'Wrong format for header "{header}"')
        return None
    token = parse_result[0]

然后，我们解码令牌。如果无法使用公钥解码令牌，则会引发 DecodeError。令牌也可以过期：

    try:
        decoded_token = decode_token(token.encode('utf8'), public_key)
    except jwt.exceptions.DecodeError:
        logger.warning(f'Error decoding header "{header}". '
        'This may be key missmatch or wrong key')
        return None
    except jwt.exceptions.ExpiredSignatureError:
        logger.info(f'Authentication header has expired')
        return None

然后，检查它是否具有预期的 exp 和 username 参数。如果缺少这些参数中的任何一个，则意味着令牌格式在解码后不正确。在不同版本中更改代码时可能会发生这种情况：

    # Check expiry is in the token
    if 'exp' not in decoded_token:
        logger.warning('Token does not have expiry (exp)')
        return None

    # Check username is in the token
    if 'username' not in decoded_token:
        logger.warning('Token does not have username')
        return None

    logger.info('Header successfully validated')
    return decoded_token['username']

如果一切顺利，最后返回用户名。

每个可能的问题都以不同的严重性记录。大多数常见事件都使用信息级安全记录，因为它们并不严重。令牌解码后出现格式错误等情况可能表明我们的编码过程存在问题。

请注意，我们使用私钥/公钥架构而不是对称密钥架构来编码和解码令牌。这意味着解码和编码密钥是不同的。

在我们的微服务结构中，只有签名权限需要私钥。这增加了安全性，因为其他服务中的任何密钥泄漏都无法检索能够签署承载令牌的密钥。不过，我们需要生成正确的私钥和公钥。

要生成私钥/公钥，请运行以下命令：

$ openssl genrsa -out key.pem 2048
Generating RSA private key, 2048 bit long modulus
.....................+++
.............................+++

然后，要提取公钥，请使用以下命令：

$ openssl rsa -in key.pem -outform PEM -pubout -out key.pub

这将生成两个文件：key.pem 和 key.pub 带有一对私钥/公钥。以文本格式读取它们就足以将它们用作编码/解码 JWT 令牌的键：

>> with open('private.pem') as fp:
>> ..  private_key = fp.read()

>> generate_token_header('peter', private_key)
'Bearer <token>'

请注意，对于测试，我们生成了一个作为字符串附加的示例密钥对。这些密钥是专门为此用途创建的，不会在其他任何地方使用。请不要在任何地方使用它们，因为它们在 GitHub 中公开可用。

为了测试我们的应用程序，我们使用了出色的 pytest 框架，它是 Python 应用程序测试运行程序的黄金标准。

基本上，pytest 有很多插件和附加组件来处理很多情况。我们将使用 pytest-flask，它有助于运行 Flask 应用程序的测试。

要运行所有测试，只需在命令行中调用 pytest：

$ pytest
============== test session starts ==============
....
==== 17 passed, 177 warnings in 1.50 seconds =====

我们配置基本用法，包括始终启用 pytest.ini 文件中的标志和 conftest.py 中的固定装置。

pytest 中使用 Fixtures 来准备应该执行测试的上下文，准备它并在最后清理它。 pytest-flask 期望应用程序夹具，如文档中所示。该插件生成一个 client 夹具，我们可以使用它在测试模式下发送请求。我们在 thoughts_fixture 夹具中看到了这个夹具，它通过 API 生成三个想法，并在我们的测试运行后删除所有内容。

简化后的结构如下：

1. Generate three thoughts. Store its thought_id:

@pytest.fixture
def thought_fixture(client):

    thought_ids = []
    for _ in range(3):
        thought = {
            'text': fake.text(240),
        }
        header = token_validation.generate_token_header(fake.name(),
                                                        PRIVATE_KEY)
        headers = {
            'Authorization': header,
        }
        response = client.post('/api/me/thoughts/', data=thought,
                               headers=headers)
        assert http.client.CREATED == response.status_code
        result = response.json
        thought_ids.append(result['id'])

2. Then, add yield thought_ids to the test:

yield thought_ids

3. Retrieve all thoughts and delete them one by one:

# Clean up all thoughts
response = client.get('/api/thoughts/')
thoughts = response.json
for thought in thoughts:
    thought_id = thought['id']
    url = f'/admin/thoughts/{thought_id}/'
    response = client.delete(url)
    assert http.client.NO_CONTENT == response.status_code

请注意，我们使用 faker 模块来生成假名称和文本。您可以在 https://faker.readthedocs.io/en/stable/ 上查看其完整文档。这是为您的测试生成随机值的好方法，可以避免重复使用 test_user 和 test_text。通过独立检查输入而不是盲目地复制占位符，它还有助于塑造您的测试。

还要注意 client 夹具的使用，它由 pytest-flask 提供。

此测试文件测试 token_validation 模块的行为。该模块涵盖了身份验证标头的生成和验证，因此彻底测试它很重要。

测试检查是否可以使用正确的密钥对标头进行编码和解码。它还检查无效输入方面的所有不同可能性：不正确格式的不同形状、无效的解码密钥或过期的令牌。

为了检查过期的令牌，我们使用两个模块：freezegun，让测试检索特定的测试时间（https://github.com/spulec/freezegun) 和 delorean，以便轻松解析日期（虽然，该模块的功能更多；查看文档https://delorean.readthedocs.io/en/latest/）。这两个模块非常易于使用，非常适合测试目的。

例如，此测试检查过期的令牌：

@freeze_time('2018-05-17 13:47:34')
def test_invalid_token_header_expired():
    expiry = delorean.parse('2018-05-17 13:47:33').datetime
    payload = {
        'username': 'tonystark',
        'exp': expiry,
    }
    token = token_validation.encode_token(payload, PRIVATE_KEY)
    token = token.decode('utf8')
    header = f'Bearer {token}'
    result = token_validation.validate_token_header(header, PUBLIC_KEY)
    assert None is result

请注意，冻结时间恰好是令牌到期时间后的 1 秒。

用于测试的公钥和私钥在 constants.py 文件中定义。如果您使用无效的公钥解码令牌，则有一个额外的独立公钥用于检查会发生什么。

该文件检查定义的 API 接口。每个 API 都经过测试以正确执行操作（创建新想法、返回用户的想法、检索所有想法、搜索想法以及通过 ID 检索想法）以及一些错误测试（创建和检索想法的未经授权的请求用户，或检索一个不存在的想法）。

在这里，我们再次使用 freezegun 来确定创建想法的时间，而不是使用取决于运行测试时间的时间戳创建它们。

在本章中，我们了解了如何开发 Web 微服务。我们首先按照 REST 原则设计其 API。然后，我们描述了如何访问数据库的模式，以及如何使用 SQLAlchemy 进行访问。

然后，我们学习了如何使用 Flask-RESTPlus 来实现它。我们学习了如何定义映射到 API 端点的资源，如何解析输入值，如何处理动作，以及如何使用序列化模型返回结果。我们描述了身份验证层的工作原理。

我们包含了测试并描述了如何使用 pytest 夹具为我们的测试创建初始条件。在下一章中，我们将了解如何将服务容器化并通过 Docker 运行它。

1. Can you name the characteristics of RESTful applications?
2. What are the advantages of using Flask-RESTPlus?
3. Which alternative frameworks to Flask-RESTPlus do you know?
4. Name the Python package used in the tests to fix the time.
5. Can you describe the authentication flow?
6. Why did we choose SQLAlchemy as a database interface for the example project?

有关不限于 Python 的 RESTful 设计的深入描述，您可以在Hands-On RESTful API Design Patterns and Best Practices（https://www.packtpub.com/gb/application -development/hands-restful-api-design-patterns-and-best-practices）。您可以在Flask：构建 Python Web 服务（https://www.packtpub.com/gb/web-development/flask-building-python-web-services)。