网站建设技术团队有多重要营销技巧五步推销法

原文：Building REST APIs with Flask

协议：CC BY-NC-SA 4.0

一、从 Flask 开始

Flask 是一个 BSD 许可的 Python 微框架，基于 Werkzeug 和 Jinja2。作为一个微框架并不会降低它的功能性；Flask 是一个非常简单但高度可扩展的框架。这使得开发人员能够选择他们想要的配置，从而使编写应用或插件变得容易。Flask 最初是由 Pocoo(一个开源开发团队)在 2010 年创建的，现在由 Pallets 项目开发和维护，该项目为 Flask 背后的所有组件提供动力。Flask 由一个活跃的、有帮助的开发者社区支持，包括一个活跃的 IRC 频道和一个邮件列表。

Flask 简介

Flask 有两个主要组件，Werkzeug 和 Jinja2。Werkzeug 负责提供路由、调试和 Web 服务器网关接口(WSGI)，而 Flask 利用 Jinja2 作为模板引擎。Flask 本身不支持数据库访问、用户认证或任何其他高级实用程序，但它支持扩展集成来添加所有这些功能，这使 Flask 成为一个用于开发 web 应用和服务的微型生产就绪框架。一个简单的 Flask 应用可以放入一个 Python 文件中，也可以模块化来创建一个生产就绪的应用。Flask 背后的想法是为所有的应用建立一个良好的基础，把其他的一切都留在扩展上。

Flask 社区非常大，并且非常活跃，有数百个开源扩展。Flask 核心团队不断审查扩展，并确保批准的扩展与未来版本兼容。Flask 作为一个微框架，为开发人员提供了选择适合他们项目的设计决策的灵活性。它维护一个定期更新和持续维护的扩展注册表。

起始 Flask

Flask 就像所有其他 Python 库一样，可以从 Python 包索引(PPI)安装，并且非常容易设置和开始开发，只需要几分钟就可以开始使用 Flask。为了能够理解这本书，你应该熟悉 Python、命令行(或者至少是 PIP)和 MySQL。

正如承诺的那样，Flask 非常容易上手，仅五行代码就可以让您开始使用一个最小的 Flask 应用。

from flask import Flask
app = Flask(__name__)@app.route('/')
def hello_world():return 'Hello, From Flask!'if __name__== '__main__':app.run()Listing 1-1Basic Flask Application

前面的代码导入 Flask 库，通过创建 Flask 类的实例启动应用，声明路由，然后定义调用路由时要执行的函数。这段代码足以启动您的第一个 Flask 应用。

下面的代码启动了一个非常简单的内置服务器，这对于测试来说已经足够好了，但是当您想要投入生产时，可能就不行了，但是我们将在后面的章节中介绍这一点。

当该应用启动时，索引路由将根据请求返回“您好，来自 Flask！”如图 1-1 所示。

图 1-1

Flask 最小应用

本书涵盖的 Flask 组件

既然已经向您介绍了 Flask，我们将讨论本书 Flask REST API 开发中涉及的组件。

这本书将作为使用 Flask 开发 REST API 的实用指南，我们将使用 MySQL 作为后端数据库。正如已经讨论过的，Flask 没有自带数据库访问支持，为了弥补这一缺陷，我们将使用一个名为 Flask-SQLAlchemy 的 Flask 扩展，它在 Flask 中增加了对 SQLAlchemy 的支持。SQLAlchemy 本质上是一个 Python SQL 工具包和对象关系映射器，它为开发人员提供了 SQL 的全部功能和灵活性。

SQLAlchemy 提供了对企业级设计模式的全面支持，是为高性能数据库访问而设计的，同时保持了效率和易用性。我们将构建一个用户认证模块，CRUD(创建、读取、更新和删除)REST APIs，用于对象创建、检索、操作和删除。我们还将集成一个名为 Swagger 的文档实用程序，用于创建 API 文档、编写单元和集成测试、学习应用调试，最后，了解在云平台上部署和监控 REST APIs 以供生产使用的不同方法。

对于单元测试，我们将使用 pytest，这是一个全功能的 Python 测试工具——pytest 易于编写测试，并且可扩展以支持复杂的用例。我们还将使用 Postman，它是一个完整的 REST API 平台——Postman 为 API 生命周期的每个阶段提供集成工具，使 API 开发更容易、更可靠。

API 部署和监控是 REST API 开发的关键部分；当谈到为生产用例扩展 API 时，开发范式发生了巨大的变化，为了本书，我们将使用 uWSGI 和 Nginx 在云 Ubuntu 服务器上部署我们的 REST APIs。我们还将在 Heroku 上部署 REST APIs，Heroku 是一个云平台，有助于 Flask 应用的部署和开箱即用。

最后但同样重要的是，我们将讨论调试常见的 Flask 错误和警告，调试 Nginx 请求，并检查 Flask 应用监控，以确保生产使用的停机时间最少。

RESTful 服务简介

表述性状态转移(REST)是 web 服务的一种软件架构风格，它为不同类型的系统之间的数据通信提供了标准。Web 服务是开放的标准 web 应用，它以交换数据为目的与其他应用进行交互，使其成为现代 web 和移动应用中客户端服务器架构的重要组成部分。简而言之，REST 是为了计算机系统之间的互操作性而在 Web 上交换数据的标准。符合 REST 架构风格的 Web 服务被称为 RESTful web 服务，它允许请求系统使用一组统一的、预定义的无状态操作来访问和操作数据。

自从 Roy Feilding 在 2000 年提出 RESTful 架构以来，RESTful 架构已经有了很大的发展，并且已经在数百万个系统中实现。REST 现在已经成为基于 web 的应用的最重要的技术之一，并且随着它在移动和基于物联网的应用中的集成，可能会增长得更快。每一种主要的开发语言都有构建 REST web 服务的框架。REST 原则是它受欢迎和被大量使用的原因。REST 是无状态的，这使得任何类型的系统都可以直接使用 REST，并且每个请求都可以由不同的系统提供服务。

REST 使我们能够区分客户机和服务器，让我们独立地实现客户机和服务器。REST 最重要的特性是它的无状态性，也就是说客户机和服务器都不需要知道对方的状态就可以进行通信。这样，客户端和服务器都可以理解接收到的任何消息，而无需查看之前的消息。既然我们在讨论 RESTful web 服务，那么让我们深入 web 服务并比较其他 web 服务标准。

简单地说，Web 服务是一种由一个电子设备向另一个电子设备提供的服务，能够通过万维网进行通信。在实践中，web 服务提供面向资源的、基于 web 的接口给数据库服务器，并由另一个 web 客户机使用。Web 服务为不同类型的系统相互通信提供了一个平台，使用一种解决方案，程序能够以它们理解的语言相互通信(图 1-2 )。

图 1-2

REST 架构图

SOAP(简单对象访问协议)是另一种 web 服务通信协议，近年来已经被 REST 取代。根据 Stormpath 的数据，REST 服务现在主导着这个行业，代表了超过 70%的公共 API。它们通过公开一致的接口来访问命名资源。然而，SOAP 将应用逻辑的组件公开为服务，而不是数据。SOAP 现在是最初由微软创建的遗留协议，与 REST 相比，它有许多其他限制。SOAP 只通过 XML 交换数据，REST 提供了通过各种数据格式交换数据的能力。RESTful 服务相对来说速度更快，资源消耗更少。然而，SOAP 仍然有自己的用例，在这些用例中，它是比 REST 更受欢迎的协议。

当健壮的安全性至关重要时，最好使用 SOAP，因为它提供了对 web 服务安全性(WS-Security)的支持，这是一种规范，定义了如何在 Web 服务中实现安全措施以保护它们免受外部攻击。SOAP 优于 REST 的另一个优点是它内置的重试逻辑，可以补偿失败的请求，这与 REST 不同，REST 中客户端必须通过重试来处理失败的请求。SOAP 与其他技术和协议(如 WS-Security、WS-addressing、WS-coordination 等)高度可扩展，这使它比其他 web 服务协议更具优势。

现在，当我们简要讨论了 web 服务——REST 和 SOAP——之后，让我们来讨论 REST 协议的特性。一般来说，REST 服务是使用以下特性定义和实现的:

1. 统一界面

2. 陈述

3. 信息

4. 资源之间的链接

5. 贮藏

6. 无国籍的

统一界面

RESTful 服务应该有一个统一的接口来访问资源，顾名思义，API 的系统接口在整个系统中应该是统一的。一个具有统一的获取和操作数据方式的逻辑 URI 系统使得 REST 易于使用。HTTP/1.1 提供了一组处理基于名词的资源的方法；为此，这些方法通常被称为动词。

在 REST 架构中，有一个安全和幂等方法的概念。安全方法是不像 GET 或 HEAD 方法那样修改资源的方法。幂等方法是一种无论执行多少次都会产生相同结果的方法。表 1-1 提供了 RESTful 服务中常用的 HTTP 动词列表。

表 1-1

RESTful 服务中常用的 HTTP 动词

|

动词

|

create, read, update, and delete

|

操作

|

安全的

|

幂等

|
| — | — | — | — | — |
| 得到 | 阅读 | 获取单个或多个资源 | 是 | 是 |
| 邮政 | 创造 | 插入新资源 | 不 | 不 |
| 放 | 更新/创建 | 插入新资源或更新现有资源 | 不 | 是 |
| 删除 | 删除 | 删除单个或多个资源 | 不 | 是 |
| 选择 | 阅读 | 列出资源上允许的操作 | 是 | 是 |
| 头 | 阅读 | 只返回响应头，不返回正文 | 是 | 是 |
| 修补 | 更新/修改 | 仅更新对资源提供的更改 | 不 | 不 |

陈述

RESTful 服务关注资源并提供对资源的访问。在 OOP 中，资源很容易被认为是一个对象。设计 RESTful 服务时要做的第一件事是识别不同的资源并确定它们之间的关系。表示是定义资源当前状态的机器可读解释。

一旦确定了资源，下一步就是表示。REST 让我们能够使用任何格式来表示系统中的资源。不像 SOAP 限制我们使用 XML 来表示数据，我们可以使用 JSON 或者 XML。通常，JSON 是表示移动或 web 客户端调用的资源的首选方法，但是 XML 可以用来表示更复杂的资源。

下面是一个用两种格式表示资源的小例子。

{"ID": "1","Name": "Building REST APIs wiith Flask","Author": "Kunal Relan","Publisher": "Apress"
}In REST Systems, you can use either of the methods or both the methods depending on the requesting client to represent the data.Listing 1-3JSON Representation of a Book resource

<?xml version="1.0" encoding="UTF-8"?>
<Book><ID> 1 </ID><Name> Building REST APIs with Flask </Name><Author> Kunal Relan </Author><Publisher > Apress </ Publisher >
</Book>Listing 1-2XML Representation of a Book Resource

信息

在 REST 架构中，消息是一个重要的关键，它本质上建立了客户机-服务器风格的数据通信方式。客户端和服务器通过消息相互通信，其中客户端向服务器发送消息，这通常被称为请求，而服务器发送响应。除了客户端和服务器之间以请求和响应主体的形式交换的实际数据之外，客户端和服务器还以请求和响应头的形式交换一些元数据。HTTP 1.1 以如下方式定义了请求和响应头格式，以便在不同种类的系统之间实现统一的数据通信方式(图 1-3 )。

图 1-3

HTTP 示例请求

在图 1-4 中，GET 是请求方法，“/comments”是服务器中的路径，“postId=1”是请求参数，“HTTP/1.1”是客户端正在请求的协议版本，“jsonplaceholder.typicode.com”是服务器主机，内容类型是请求头的一部分。所有这些结合起来就是服务器能够理解的 HTTP 请求。

作为回报，HTTP 服务器发送对所请求资源的响应。

[{"postId": 1,"id": 1,"name": "id labore ex et quam laborum","email": "Eliseo@gardner.biz","body": "laudantium enim quasi est quidem magnam voluptate ipsam eos\ntempora quo necessitatibus\ndolor quam autem quasi\nreiciendis et nam sapiente accusantium"},{"postId": 1,"id": 2,"name": "quo vero reiciendis velit similique earum","email": "Jayne_Kuhic@sydney.com","body": "est natus enim nihil est dolore omnis voluptatem numquam\net omnis occaecati quod ullam at\nvoluptatem error expedita pariatur\nnihil sint nostrum voluptatem reiciendis et"},{"postId": 1,"id": 3,"name": "odio adipisci rerum aut animi","email": "Nikita@garfield.biz","body": "quia molestiae reprehenderit quasi aspernatur\naut expedita occaecati aliquam eveniet laudantium\nomnis quibusdamdelectus saepe quia accusamus maiores nam est\ncum et ducimus et vero voluptates excepturi deleniti ratione"},{"postId": 1,"id": 4,"name": "alias odio sit","email": "Lew@alysha.tv","body": "non et atque\noccaecati deserunt quas accusantium unde odit nobis qui voluptatem\nquia voluptas consequuntur itaque dolor\net qui rerum deleniti ut occaecati"},{"postId": 1,"id": 5,"name": "vero eaque aliquid doloribus et culpa","email": "Hayden@althea.biz","body": "harum non quasi et ratione\ntempore iure ex voluptates in ratione\nharum architecto fugit inventore cupiditate\nvoluptates magni quo et"}]

图 1-4

HTTP 示例响应

在上图中，“HTTP/2”是响应 HTTP 版本，“200”是响应代码。“cf-ray”下面的部分是响应头，“cf-ray”下面的 post 注释数组是请求的响应体。

资源之间的链接

资源是 REST 架构世界中的基本概念。资源是一个具有类型、相关数据、与其他资源的关系以及一组可以在其上执行的方法的对象。REST API 中的资源可以包含到应该驱动流程流的其他资源的链接。例如在 HTML 网页中，主页中的链接驱动用户流，REST API 中的资源应该能够在用户不知道流程图的情况下驱动用户流。

{"ID": "1","Name": "Building REST APIs wiith Flask","Author": "Kunal Relan","Publisher": "Apress","URI" : "https://apress.com/us/book/123456789"
}Listing 1-4A Book with Link to Buy

贮藏

缓存是一种技术，它存储给定资源的副本，并在请求时返回，从而节省额外的数据库调用和处理时间。它可以在不同的层次上完成，如客户机、服务器或中间件代理服务器。缓存是提高 API 性能和扩展应用的重要工具；但是，如果管理不当，会导致向客户端提供旧的结果。REST APIs 中的缓存是使用 HTTP 头来控制的。缓存头是 HTTP 头规范的重要组成部分，也是高效扩展 web 服务的重要组成部分。在 REST 规范中，当在资源 URL 上使用安全方法时，反向代理通常会缓存结果，以便在下次请求相同的资源时使用缓存的数据。

无国籍的

从客户端到服务器的每个请求都必须包含理解请求所需的所有信息，并且不能利用服务器上存储的任何上下文。因此，会话状态完全保存在客户端

—罗伊·菲尔丁

这里的无状态意味着每个 HTTP 响应本身是一个完整的实体，足以提供要执行的信息，而不需要另一个 HTTP 请求。无状态的意义在于破坏了与服务器保持一致的目的，即允许基础设施具有预期的灵活性。为了方便起见，REST 服务器在 HTTP 响应中提供了客户机可能需要的足够信息。无状态是能够扩展基础设施的重要部分，使我们能够部署多个服务器来服务数百万并发用户，因为不存在服务器会话状态依赖性。它还支持 REST 基础设施的缓存特性，因为它让缓存服务器决定是否缓存请求，只需查看特定的请求，而不考虑之前的任何请求。

规划 REST API

下面是我们在计划创建 REST APIs 时需要检查的事项列表:

1. 理解用例。知道你为什么要构建 API 以及 API 将提供什么服务是非常重要的。

2. 列出 API 特性来理解你的 API 将要做的所有动作。这还包括列出行动，并将它们组合在一起，以处理冗余的端点。

3. 确定将使用 API 的不同平台，并相应地提供支持。

4. 对支持增长和扩展基础架构进行长期规划。

5. 规划 API 版本控制策略，确保对不同版本的 API 提供持续支持。

6. 规划 API 访问策略，即身份验证、ACL 和限制。

7. 计划 API 文档和测试。

8. 理解如何在你的 API 中使用超媒体。

因此，这是在规划你的 API 时需要确保的八件重要的事情，并且对于开发一个稳定的、以生产为中心的 API 系统是非常关键的。

API 设计

现在让我们来看看 API 设计。在这里，我们将讨论设计 REST APIs 的标准，记住我们刚刚谈到的一系列事情。

长期实施

长期实施帮助你在实际实施前分析设计上的缺陷。这有助于开发人员选择合适的平台和工具，以确保相同的系统以后可以扩展到更多的用户。

规格驱动开发

规范驱动的开发使用定义而不仅仅是代码来实施 API 设计，这确保了在 API 设计完好无损的情况下对代码库进行更改。使用像 API Designer 这样的工具在开发之前理解 API 设计是一个很好的实践，这也可以让你预见缺陷。像 swagger 或 RAML 这样的工具可以让您保持 API 设计的标准化，并在需要时将 API 移植到不同的平台上。

样机研究

一旦 API 规范到位，原型设计就可以让开发人员创建模拟 API，帮助他们理解 API 的每个潜在方面，从而帮助您在实际开发之前将 API 可视化。

认证和授权

身份验证涉及到知道这个人是谁的验证过程，但它还不涉及授予对所有资源的访问权，这就是授权的由来，它涉及到授权经过身份验证的人使用访问控制列表(ACL)检查允许访问的资源。

我们有不同的用户认证和授权方式，如基本认证、HMAC 和 OAuth。然而，OAuth 2.0 是实现这一点的首选方法，并且是企业和小公司在其 REST APIs 中用于身份验证和授权的标准协议。

所以，这些是 REST 基础设施的关键特性，我们将在后面的章节中更多地讨论 REST 是如何工作的以及如何实现更好的通信。

现在，我们将开始设置我们的开发环境，并了解使用 Python 开发应用的一些关键因素。

设置开发环境

在这一部分中，我们将讨论为 Flask 应用设置 Python 开发环境。我们将使用虚拟环境作为我们的依赖项的独立隔离环境。在设置开发环境的过程中，我们将使用 PIP 来安装和管理我们的依赖项以及一些其他有用的实用程序。为了这本书，我们将在 macOS Mojave 和 Python 2.7 上做所有的事情，但是你可以根据自己的方便随意使用任何操作系统。因此，如果您的操作系统中没有安装 Python 的正确版本，您可以使用此链接继续在您选择的操作系统上安装 Python:www.python.org/downloads/(图 1-5 )。

图 1-5

Python 下载

使用 PIP

PIP 是 PyPi 推荐的项目依赖管理工具。如果您使用从 www.python.org 下载的 Python，PIP 会预装 Python。

但是，如果您的系统中没有安装 PIP，请按照此处的指南安装 PIP。

要安装 PIP，请在终端中使用以下命令(或 Windows 中的命令行)下载 get-pip.py。

$ curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py

获得 get-pip.py 文件后，安装并运行下一个命令:

$ python get-pip.py

前面的命令将安装 PIP、setuptools(安装源发行版所需的)和 wheel。

如果您已经有 pip，您可以使用以下命令升级到 PIP 的最新版本:

$ pip install -U pip

要测试您的安装，您应该在您的终端(或 Windows 中的命令行)中运行以下命令(图 1-6 ):

图 1-6

检查 Python 和 PIP 安装

$ python -V
$ pip -V

选择 IDE

在我们开始写代码之前，我们需要一些东西来写。在本书中，我们将使用 Visual Studio 代码，这是一个在所有主要操作系统上都可用的开源免费 IDE。Visual Studio 代码可以从 www.code.visualstudio.com 下载，它为开发 Python 应用提供了很好的支持，提供了大量方便的插件来方便开发。您可以选择使用自己喜欢的文本编辑器或 IDE 来阅读这本书(图 1-7 )。

图 1-7

Visual Studio 代码

一旦我们有了 IDE 设置，我们就可以开始安装和设置虚拟环境了。

了解 Python 虚拟环境

Python 就像其他现代编程语言一样，提供了大量的第三方库和 SDK。不同的应用可能需要各种特定版本的第三方模块，一个 Python 安装不可能满足每个应用的这种需求。因此，在 Python 的世界中，这个问题的解决方案是虚拟环境，它创建一个独立的自包含目录树，包含所需版本的 Python 安装以及所需的包。

从本质上讲，虚拟环境的主要目的是创建一个隔离的环境来包含 Python 的安装和应用所需的包。您可以创建的虚拟环境的数量没有限制，并且创建它们非常容易。

使用虚拟环境

在 Python 2.7 中，我们需要一个名为 virtualenv 的模块，它是使用 PIP 安装的，以便开始使用 Python 虚拟环境。

注意

在 Python 3 中，venv 模块是作为标准库的一部分预装的。

要安装 virtualenv，请在终端中键入以下命令(如果是 Windows，请键入命令行)。

$ pip install virtualenv

一旦我们在系统中安装了 virtualenv 模块，接下来我们将创建一个新目录，并在其中创建一个虚拟环境。

现在，键入以下命令创建一个新目录，并在终端中打开它。

$ mkdir pyenv && cd pyenv

前面的命令将创建一个目录，并在您的终端中打开它，然后我们将使用 virtualenv 模块在目录中创建一个新的虚拟环境。

$ virtualenv venv

前面的命令将使用 virtualenv 模块并创建一个名为 venv 的虚拟环境。你可以给你的虚拟环境起任何名字，但是在这本书里，为了统一起见，我们只使用 venv。

一旦这个命令停止执行，您将看到一个名为 venv 的目录。这个目录现在将保存您的虚拟环境。

venv 文件夹的目录结构应类似于图 1-8 中的目录结构。

图 1-8

虚拟环境目录结构

下面是结构中每个文件夹包含的内容:

1. bin:与虚拟环境交互的文件。

2. include: C 头文件来编译 Python 包。

3. lib:这个文件夹包含 Python 版本和所有其他第三方模块的副本。

接下来，有不同 Python 工具的副本或符号链接，以确保所有 Python 代码和命令都在当前环境中执行。这里重要的部分是 bin 文件夹中的激活脚本，它将 shell 设置为使用虚拟环境的 Python 和 site 包。为此，您需要通过在终端中键入以下命令来激活虚拟环境。

$ source venv/bin/activate

一旦执行了这个命令，您的 shell 提示符将会以虚拟环境的名称为前缀，如图 1-9 所示。

图 1-9

激活虚拟环境

现在，让我们使用以下命令在我们的虚拟环境中安装 Flask:

$ pip install flask

前面的命令应该在我们的虚拟环境中安装 Flask。我们将使用我们在示例 Flask 应用中使用的相同代码。

$ nano app.py

并在 nano 文本编辑器中键入以下代码:

from flask import Flask
app = Flask(__name__)@app.route('/')
def hello_world():return 'Hello, From Flask!'Now,  try running your app.py using python app.py command.$ FLASK_APP=app.py flask run

使用前面的命令，您应该能够运行简单的 Flask 应用，并且应该在您的终端中看到类似的输出(图 1-10 )。

图 1-10

在虚拟环境中运行 Flask 应用

现在，要停用虚拟环境，您需要执行以下命令:

$ deactivate

该命令执行后，shell 中的(venv)前缀将消失，如果您尝试再次运行该应用，它将抛出一个错误(图 1-11 )。

图 1-11

不使用虚拟环境运行 Flask 应用

现在，您已经了解了虚拟环境的概念，我们可以更深入地了解虚拟环境内部的情况。

理解虚拟环境的工作方式可以真正帮助您调试应用和理解执行环境。首先，让我们检查一下激活和停用虚拟环境的 Python 可执行文件，以便理解基本的区别。

让我们在激活虚拟环境的情况下执行以下命令(图 1-12 ):

图 1-12

使用虚拟环境检查 Python 可执行文件

$ which python

如下图所示，外壳正在使用虚拟环境的 Python 可执行文件，如果您停用该环境并重新运行 Python 命令，您会注意到外壳现在正在使用系统的 Python(图 1-13 )。

图 1-13

在没有虚拟环境的情况下检查 Python 可执行文件

因此，一旦激活了虚拟环境，就会修改$path 环境变量，使其指向我们的虚拟环境，从而使用我们虚拟环境中的 Python，而不是系统环境。然而，这里需要注意的一件重要的事情是，它基本上是系统的 Python 可执行文件的一个副本或一个符号链接。

安装 Flask

我们已经在前面的模块中安装了 Flask，但是让我们重新开始并设置 Flask 微框架。

安装 Flask

激活虚拟环境后，执行以下命令安装最新版本的 Flask。

$pip install flask

前面的命令将在您的虚拟环境中安装 Flask。

但是，如果您希望在发布之前使用最新的 Flask，请通过执行以下命令，使用其存储库的主分支安装/更新 Flask 模块:

$pip install -U https://github.com/pallets/flask/archive/master.tar.gz

当您安装 Flask 时，以下发行版会随主框架一起安装:

1. Werkzeug(http://werkzeug.pocoo.org/*):*Werkzeug 实现了 WSGI，应用和服务器之间的标准 Python 接口。

2. Jinja ( http://jinja.pocoo.org/ ): Jinja 是 Flask 中的模板引擎，为应用呈现页面。

3. MarkupSafe (

4. its dangerous(https://pythonhosted.org/itsdangerous/):its dangerous 负责安全地对数据进行签名，以确保数据的完整性，并用于保护 Flask 会话 cookies。

5. Click ( http://click.pocoo.org/ ): Click 是一个写 CLI 应用的框架。它提供了“Flask”CLI 命令。

结论

一旦在虚拟环境中安装了 Flask，就可以进入开发阶段的下一步了。在此之前，我们将讨论 MySQL 和 Flask-SQLAlchemy，它是我们将在 Flask 应用中使用的 ORM。数据库是 REST 应用的重要组成部分，在下一章，我们将讨论 MySQL 数据库和 Flask-SQLAlchemy ORM，并学习如何将我们的 Flask 应用与 Flask-SQLAlchemy 连接起来。

二、Flask 中的数据库建模

本章涵盖了 REST 应用开发的一个最重要的方面，即与数据库系统的连接和交互。在本章中，我们将讨论 NoSQL 和 SQL 数据库，以及它们之间的连接和交互。

在本章中，我们将讨论以下主题:

1. NoSQL 与 SQL 数据库

2. 连接 Flask-SQLAlchemy

3. 使用 Flask-SQLAlchemy 与 MySQL 数据库交互

4. 连接 Flask-MongoEngine

5. 使用 Flask-MongoEngine 与 MongoDB 交互

介绍

Flask 作为一个微框架，为应用提供了数据源的灵活性，并为与不同种类的数据源进行交互提供了库支持。Flask 中有一些库可以连接到基于 SQL 和基于 NoSQL 的数据库。它还提供了使用原始数据库库或使用 ORM(对象关系映射器)/ODM(对象文档映射器)与数据库进行交互的灵活性。在这一章中，我们将简要讨论基于 NoSQL 和 SQL 的数据库，并通过 Flask-SQLAlchemy 学习如何在 Flask 应用中使用 ORM 层，之后我们将通过 Flask-MongoEngine 使用 ODM 层。

大多数应用在某些时候确实需要数据库，MySQL 和 MongoDB 只是众多工具中的两个。为您的应用选择正确的方法完全取决于您要存储的数据。如果表中的数据集是相互关联的，那么 SQL 数据库是一个不错的选择，或者 NoSQL 数据库也可以达到这个目的。

现在，让我们简要地看一下 SQL 和 NoSQL 数据库。

SQL 数据库

SQL 数据库使用结构化查询语言(SQL)进行数据操作和定义。SQL 是一个通用的、被广泛使用和接受的选项，这使它成为数据存储的最佳选择。当使用的数据需要是关系型的并且模式是预定义的时，SQL 系统非常适合。然而，预定义的模式也有缺点，因为它要求整个数据集遵循相同的结构，这在某些情况下可能很困难。SQL 数据库以由行和列组成的表格的形式存储数据，并且是垂直可伸缩的。

NoSQL 数据库

NoSQL 数据库有一个非结构化数据的动态模式，并以不同的方式存储数据，包括基于列的(Apache Cassandra)、基于文档的(MongoDB)和基于图形的(Neo4J)或作为键值存储(Redis)。这提供了在没有预定义结构的情况下存储数据的灵活性，以及随时向数据结构添加字段的多功能性。无模式是 NoSQL 数据库的主要特点，这也使它们更适合分布式系统。与 SQL 数据库不同，NoSQL 数据库是水平可伸缩的。

既然我们已经简要解释了 SQL 和 NoSQL 数据库，我们将跳转到 MySQL 和 MongoDB 之间的功能差异，因为这是我们将在本章中研究的两个数据库引擎。

主要区别:MySQL 与 MongoDB

如前所述，MySQL 是一个基于 SQL 的数据库，它将数据存储在具有列和行的表中，并且只处理结构化数据。另一方面，MongoDB 可以处理非结构化数据，存储类似 JSON 的文档而不是表格，并使用 MongoDB 查询语言与数据库通信。MySQL 是一个非常成熟的数据库，具有巨大的社区和极大的稳定性，而 MongoDB 是一项相当新的技术，社区不断增长，由 MongoDB Inc .开发。MySQL 是垂直可伸缩的，其中单个服务器上的负载可以通过升级 RAM、SSD 或 CPU 来增加，而在 MongoDB 的情况下，它需要共享和添加更多的服务器，以便增加服务器负载。MongoDB 是高写负载和大数据集的首选，MySQL 非常适合高度依赖多行事务的应用，如会计系统。对于具有动态结构和高数据负载的应用，如实时分析应用或内容管理系统，MongoDB 是一个很好的选择。

Flask 提供了与 MySQL 和 MongoDB 交互的支持。有各种本地驱动程序以及 ORM/ODM 用于与数据库通信。MySQL 是一个 Flask 扩展，允许本地连接到 MySQL；PyMongo 是在 Flask 中使用 MongoDB 的本地扩展，也是 MongoDB 推荐的。Flask-MongoEngine 是一个 Flask 扩展，用于 Flask 和 MongoDB 的 ODM。Flask-SQLAlchemy 是一个 ORM 层，用于 Flask 应用连接 MySQL。

接下来，我们将讨论 Flask-SQLAlchemy 和 Flask- MongoEngine，并使用它们创建 Flask CRUD 应用。

使用 SQLAlchemy 创建 Flask 应用

Flask-SQLAlchemy 是 Flask 的扩展，为应用增加了对 SQLAlchemy 的支持。SQLAlchemy 是一个 Python 工具包和对象关系映射器，使用 Python 提供对 SQL 数据库的访问。SQLAlchemy 提供了企业级的持久性模式和高效高性能的数据库访问。如果安装了适当的 DBAPI 驱动程序，Flask-SQLAlchemy 支持以下基于 SQL 的数据库引擎:

• 一种数据库系统

• 关系型数据库

• 神谕

• 数据库

• 搜寻配置不当的

• 火鸟赛贝斯

我们将在应用中使用 MySQL 作为数据库引擎，所以让我们开始安装 SQLAlchemy 并设置我们的应用。

让我们创建一个名为 flask-MySQL 的新目录，创建一个虚拟环境，然后安装 flask-sqlalchemy。

$ mkdir flask-mysql && cd flask-mysql

现在，使用以下命令在目录中创建一个虚拟环境:

$ virtualenv venv

如前所述，我们可以使用以下命令激活虚拟环境:

$ source venv/bin/activate

一旦虚拟环境被激活，让我们安装 flask-sqlalchemy。

Flask 和 Flask-SQLAlchemy 可以使用 PIP 和以下命令进行安装。

(venv)$ pip install flask flask-sqlalchemy

除了 SQLite，所有其他数据库引擎都需要单独的库与 Flask-SQLAlchemy 一起安装才能运行。SQLAlchemy 使用 MySQL-Python 作为与 MySQL 连接的默认 DBAPI。

现在，让我们安装 PyMySQL 来启用 MySQL 与 Flask-SQLAlchemy 的连接。

(venv) $ pip install pymysql

现在，我们应该拥有了创建示例 flask-MySQL 应用所需的一切。

让我们从创建 app.py 开始，它将包含我们的应用的代码。创建文件后，我们将启动 Flask 应用。

from flask import Flask
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://<mysql_username>:<mysql_password>@<mysql_host>:<mysql_port>/<mysql_db>'
db = SQLAlchemy(app)if __name__ == "__main__":app.run(debug=True)

这里，我们导入 Flask 框架和 Flask-SQLAlchemy，然后初始化 Flask 的一个实例。之后，我们配置 SQLAlchemy 数据库 URI 以使用我们的 MySQL 数据库 URI，然后我们创建一个名为 DB 的 SQLAlchemy 对象，它将处理我们的 ORM 相关活动。

现在，如果您正在使用 MySQL，请确保您提供了正在运行的 MySQL 服务器的连接字符串，并且所提供的数据库名称确实存在。

注意

使用环境变量在应用中提供数据库连接字符串。

确保您有一个正在运行的 MySQL 服务器来跟踪这个应用。但是，您也可以通过在 SQLAlchemy 数据库 URI 中提供 SQLite 配置详细信息来使用 SQLite，如下所示:

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:tmp/<db_name>.db'

为了运行应用，您需要在终端中执行以下代码:

(venv) $ python app.py

如果没有错误，您应该在终端中看到类似的输出:

(venv) $ python app.py
* Serving Flask app "app" (lazy loading)* Environment: productionWARNING: Do not use the development server in a production environment.Use a production WSGI server instead.* Debug mode: on* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)* Restarting with stat* Debugger is active!* Debugger PIN: 779-301-240

创建作者数据库

我们现在将创建一个作者数据库应用，它将提供 RESTful CRUD APIs。所有作者都将存储在名为“authors”的表中。

在声明的 db 对象之后，添加以下代码行，以将一个类声明为 Authors，该类将保存 author 表的模式:

class Author (db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(20))specialisation = db.Column(db.String(50))def __init__(self, name, specialisation):self.name = nameself.specialisation = specialisationdef __repr__(self):return '<Product %d>' % self.id
db.create_all()

使用这段代码，我们创建了一个名为“Authors”的模型，它有三个字段——ID、name 和 specialisation。Name 和 specialisation 是字符串，但是 ID 是一个自动生成并自动递增的整数，它将作为主键。注意最后一行“db.create_all()”，它指示应用创建应用中指定的所有表和数据库。

为了使用 SQLAlchemy 返回的数据为来自 API 的 JSON 响应提供服务，我们需要另一个名为 marshmallow 的库，它是 SQLAlchemy 的附加组件，用于将 SQLAlchemy 返回的数据对象序列化到 JSON。

(venv)$ pip install flask-marshmallow

以下命令将在我们的应用中安装 marshmallow 的 Flask 版本，我们将使用 marshmallow 从 Authors 模型中定义输出模式。

在应用文件的顶部，其他导入的下面添加下面几行来导入 marshmallow。

from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields

在 db.create_all()之后，使用以下代码定义输出模式:

class AuthorSchema(ModelSchema):class Meta(ModelSchema.Meta):model = Authorssqla_session = db.sessionid = fields.Number(dump_only=True)name = fields.String(required=True)specialisation = fields.String(required=True)

前面的代码将变量属性映射到字段对象，在 Meta 中，我们定义了与我们的模式相关的模型。所以这应该有助于我们从 SQLAlchemy 返回 JSON。

在建立了模型和返回模式之后，我们可以开始创建端点了。让我们创建第一个 GET /authors 端点来返回所有注册的作者。这个端点将查询 Authors 模型中的所有对象，并以 JSON 的形式返回给用户。但是在我们编写端点之前，将第一个导入行编辑为以下内容，以便从 Flask 导入 jsonify、make_response 和 request。

from flask import Flask, request, jsonify, make_response

在 AuthorSchema 之后，用以下代码编写第一个端点/作者:

@app.route('/authors', methods = ['GET'])
def index():get_authors = Authors.query.all()author_schema = AuthorSchema(many=True)authors, error = author_schema.dump(get_authors)return make_response(jsonify({"authors": authors}))

在这个方法中，我们获取数据库中的所有作者，将其转储到 AuthorSchema 中，并在 JSON 中返回结果。

如果您现在启动应用并点击端点，它将返回一个空数组，因为我们还没有在 DB 中添加任何东西，但是让我们继续尝试端点。

使用 Python app.py 运行应用，然后使用首选的 REST 客户端查询端点。我将使用 Postman 来请求端点。

所以只需打开你的 Postman，让http://localhost:5000/authors查询端点(图 2-1 )。

图 2-1

获得/作者回应

您应该在您的 Postman 客户端中看到类似的结果。现在让我们创建 POST 端点，将作者添加到我们的数据库中。

我们可以通过在我们的方法中直接创建一个 Authors 类，或者通过创建一个 classMethod 在 Authors 类中创建一个新对象，然后在我们的端点中调用该方法，来将一个对象添加到表中。让我们在 Authors 类中添加 class 方法来创建一个新对象。

在字段定义之后，在 Authors 类中添加以下代码:

    def create(self):db.session.add(self)db.session.commit()return self

前面的方法使用数据创建一个新的对象，然后返回创建的对象。现在，您的 Authors 类应该如下所示:

class Authors(db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(20))specialisation = db.Column(db.String(50))def create(self):db.session.add(self)db.session.commit()return selfdef __init__(self, name, specialisation):self.name = nameself.specialisation = specialisationdef __repr__(self):return '<Author %d>' % self.id

现在，我们将创建 POST authors 端点，并在 GET 端点后编写以下代码:

@app.route('/authors', methods = ['POST'])
def create_author():data = request.get_json()author_schema = AuthorsSchema()author, error = author_schema.load(data)result = author_schema.dump(author.create()).datareturn make_response(jsonify({"author": authors}),201)

前面的方法将获取 JSON 请求数据，将数据加载到 marshmallow 模式中，然后调用我们在 Authors 类中创建的 create 方法，该方法将返回带有 201 状态代码的已创建对象。

因此，让我们用示例数据请求 POST 端点并检查响应。让我们用 JSON 请求体打开 Postman 和 POST /authors。我们需要在主体中添加名称和专门化字段来创建对象。我们的示例请求体应该如下所示:

{"name" : "Kunal Relan","specialisation" : "Python"
}

一旦我们请求了端点，我们将获得 Author 对象作为对我们新创建的 Author 的响应。注意，在这种情况下，返回状态代码是 201，这是新对象的状态代码(图 2-2 )。

图 2-2

帖子/作者端点

所以现在，如果我们请求 GET /authors 端点，我们将在响应中获得新创建的作者。

重新访问 Postman 中的 GET /authors 选项卡并再次点击请求；这一次，您应该得到一个由我们新创建的作者组成的作者数组(图 2-3 )。

图 2-3

用新对象获取所有作者

到目前为止，我们已经创建了注册新作者和获取作者列表的端点。接下来，我们将创建一个端点来使用作者 ID 返回作者，然后更新端点来使用作者 ID 更新作者详细信息，最后一个端点使用作者 ID 删除作者。

对于通过 ID 获取作者，我们将有一个类似/authors/ 的路径，它将从请求参数中获取作者 ID 并找到匹配的作者。

将以下代码添加到 GET author by ID 端点的 GET all authors 路径下。

@app.route('/authors/<id>', methods = ['GET'])
def get_author_by_id(id):get_author = Authors.query.get(id)author_schema = AuthorsSchema()author, error = author_schema.dump(get_author)return make_response(jsonify({"author": author}))

接下来我们需要测试这个端点，我们将请求 ID 为 1 的 author，就像我们在前面的 GET all authors API response 中看到的那样，所以让我们再次打开 Postman 并请求应用服务器上的/authors/1 来检查响应。

图 2-4

通过 ID 端点获取作者

正如您在前面的屏幕截图中看到的，我们正在返回一个具有密钥 author 的对象，该对象包含 ID 为 1 的 author 对象。现在，您可以使用 POST 端点添加更多作者，并使用返回的 ID 获取他们。

接下来，我们需要创建一个端点来更新作者姓名或专业，为了更新任何对象，我们将使用在“RESTful 服务简介”一节中讨论过的 PUT HTTP 动词。这个端点将类似于 GET authors by ID 端点，但是将使用 PUT 动词而不是 GET 动词。

下面是 PUT 端点更新 author 对象的代码

@app.route('/authors/<id>', methods = ['PUT'])
def update_author_by_id(id):data = request.get_json()get_author = Authors.query.get(id)if data.get('specialisation'):get_author.specialisation = data['specialisation']if data.get('name'):get_author.name = data['name']db.session.add(get_author)db.session.commit()author_schema = AuthorsSchema(only=['id', 'name', 'specialisation'])author, error = author_schema.dump(get_author)return make_response(jsonify({"author": author}))

因此，让我们测试我们的 PUT 端点，并更改 author ID 1 的专门化。

我们将在下面的 JSON 主体中更新作者专门化。

{"specialisation" : "Python Applications"
}

图 2-5

按 ID 端点更新作者

如图 2-5 所示，我们用 ID 1 更新了作者，现在专门化已经更新为“Python 应用”。

现在，从数据库中删除作者的最后一个端点。添加以下代码以添加一个删除端点，该端点类似于 get author by ID endpoint，但将使用删除谓词并返回没有内容的 204 状态代码。

@app.route('/authors/<id>', methods = ['DELETE'])
def delete_author_by_id(id):get_author = Authors.query.get(id)db.session.delete(get_author)db.session.commit()return make_response("",204)

现在我们将请求删除端点删除 ID 为 1 的作者(图 2-6 )。

图 2-6

按 ID 删除作者

现在，如果您请求获取所有作者端点，它将返回一个空数组。

现在，您的 app.py 应该具有以下代码:

from flask import Flask, request, jsonify, make_response
from flask_sqlalchemy import SQLAlchemy
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fieldsapp = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = 'mysql+pymysql://<mysql_username>:<mysql_password>@<mysql_host>:<mysql_port>/<mysql_db>'db = SQLAlchemy(app)class Authors(db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(20))specialisation = db.Column(db.String(50))def create(self):db.session.add(self)db.session.commit()return selfdef __init__(self, name, specialisation):self.name = nameself.specialisation = specialisationdef __repr__(self):return '<Author %d>' % self.iddb.create_all()class AuthorsSchema(ModelSchema):class Meta(ModelSchema.Meta):model = Authorssqla_session = db.sessionid = fields.Number(dump_only=True)name = fields.String(required=True)specialisation = fields.String(required=True)@app.route('/authors', methods = ['GET'])
def index():get_authors = Authors.query.all()author_schema = AuthorsSchema(many=True)authors, error = author_schema.dump(get_authors)return make_response(jsonify({"authors": authors}))@app.route('/authors/<id>', methods = ['GET'])
def get_author_by_id(id):get_author = Authors.query.get(id)author_schema = AuthorsSchema()author, error = author_schema.dump(get_author)return make_response(jsonify({"author": author}))@app.route('/authors/<id>', methods = ['PUT'])
def update_author_by_id(id):data = request.get_json()get_author = Authors.query.get(id)if data.get('specialisation'):get_author.specialisation = data['specialisation']if data.get('name'):get_author.name = data['name']db.session.add(get_author)db.session.commit()author_schema = AuthorsSchema(only=['id', 'name', 'specialisation'])author, error = author_schema.dump(get_author)return make_response(jsonify({"author": author}))@app.route('/authors/<id>', methods = ['DELETE'])
def delete_author_by_id(id):get_author = Authors.query.get(id)db.session.delete(get_author)db.session.commit()return make_response("",204)@app.route('/authors', methods = ['POST'])
def create_author():data = request.get_json()author_schema = AuthorsSchema()author, error = author_schema.load(data)result = author_schema.dump(author.create()).datareturn make_response(jsonify({"author": result}),200)if __name__ == "__main__":app.run(debug=True)

因此，我们现在已经创建并测试了我们的样本 Flask-MySQL CRUD 应用。我们将在后面的章节中使用 Flask-SQLAlchemy 检查复杂的对象关系，接下来我们将使用 MongoEngine 创建一个类似的 Flask CRUD 应用。

示例 Flask MongoEngine 应用

正如我们所讨论的，MongoDB 是一个强大的基于文档的 NoSQL 数据库。它使用类似 JSON 的文档模式结构，具有高度的可伸缩性。在这个例子中，我们将再次创建一个 Authors 数据库 CRUD 应用，但是这次我们将使用 MongoEngine 而不是 SQLAlchemy。MongoEngine 增加了对 Flask 的 MongoDB 支持，与 SQLAlchemy 非常相似，但是它缺少一些特性，因为 MongoDB 仍然没有广泛用于 Flask。

让我们开始为 flask-mongodb 应用设置项目。就像上次一样，创建一个新的目录 flask-mongodb，并在其中初始化一个新的虚拟环境。

$ mkdir flask-mongodb && cd flask-mongodb

创建目录后，让我们生成虚拟环境并激活它。

$ virtualenv venv
$ source venv/bin/activate

现在让我们使用 PIP 安装我们的项目依赖项。

(venv) $ pip install flask

我们需要 Flask-MongoEngine 和 Flask-marshmallow，所以让我们也安装它们。

(venv) $ pip install flask-mongoengine
(venv) $ pip install flask-marshmallow

安装完依赖项后，我们可以创建 app.py 文件并开始编写代码。

因此，下面的代码是应用的框架，其中导入 flask，创建一个应用实例，然后导入 MongoEngine 创建一个 db 实例。

from flask import Flask, request, jsonify, make_response
from flask_mongoengine import MongoEngine
from marshmallow import Schema, fields, post_load
from bson import ObjectIdapp = Flask(__name__)
app.config['MONGODB_DB'] = 'authors'
db = MongoEngine(app)Schema.TYPE_MAPPING[ObjectId] = fields.Stringif __name__ == "__main__":app.run(debug=True)Here TYPE_MAAPPING helps marshmallow understand the ObjectId type while serializing and de-serializing the data.

注意

这里我们不需要 db.create_all()，因为 MongoDB 会在您第一次将值保存到集合中时动态创建它。

如果您现在运行应用，您的服务器应该会启动，但是它没有什么要处理的，只是创建 db 实例并建立连接。接下来，让我们使用 MongoEngine 创建一个作者模型。

在这种情况下，创建作者模型的代码相当简单，如下所示:

class Authors(db.Document):name = db.StringField()specialisation = db.StringField()

现在让我们创建 marshmallow 模式，我们需要用它将 db 对象转储到序列化的 JSON 中。

class AuthorsSchema(Schema):name = fields.String(required=True)specialisation = fields.String(required=True)

前面的代码让我们创建一个模式，我们将使用该模式将 db 对象映射到 marshmallow。请注意，这里我们没有使用 marshmallow-sqlalchemy，它对 sqlalchemy 有一个额外的支持层，因此代码看起来略有变化。

现在，我们可以编写 GET 端点来从数据库中获取所有作者。

@app.route('/authors', methods = ['GET'])
def index():get_authors = Authors.objects.all()author_schema =AuthorsSchema(many=True,only=['id','name','specialisation'])authors, error = author_schema.dump(get_authors)return make_response(jsonify({"authors": authors}))

注意

MongoEngine 在“Id”字段中返回唯一的 ObjectId，它是自动生成的，因此没有在模式中指定。

现在，让我们使用下面的命令再次启动应用。

(venv) $ python app.py

如果没有错误，您应该会看到下面的输出，并且您的应用应该已经启动并正在运行。

(venv) $ python app.py
* Serving Flask app "app" (lazy loading)* Environment: productionWARNING: Do not use the development server in a production environment.Use a production WSGI server instead.* Debug mode: on* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)* Restarting with stat* Debugger is active!* Debugger PIN: 779-301-240

图 2-7

请求获取/作者

既然我们的 GET 端点正在工作(图 2-7 )，让我们创建一个 POST /authors 端点来在数据库中注册作者。

@app.route('/authors', methods = ['POST'])
def create_author():data = request.get_json()author = Authors(name=data['name'],specialisation=data['specialisation'])author.save()author_schema = AuthorsSchema(only=['name', 'specialisation'])authors, error = author_schema.dump(author)return make_response(jsonify({"author": authors}),201)

前面的代码将请求 JSON 数据放在数据变量中，创建一个类 Authors 对象，并对其调用 save()方法。接下来，它使用 AuthorsSchema 创建一个模式，并转储新对象以将其返回给用户，确认用户是用 201 状态代码创建的。

现在重新运行应用，并使用示例作者详细信息请求 POST 端点进行注册。

我们将使用相同的 JSON 数据发送到这个应用，就像我们在另一个应用中所做的那样。

{"name" : "Kunal Relan","specialisation" : "Python"
}

图 2-8

请求帖子/作者

在请求时，您应该得到类似于图 2-8 中所示的输出，现在只是为了确认我们的 get 端点工作正常，我们将再次请求它，看看它是否返回数据。

如图 2-9 所示，我们在 GET /authors 端点中获得了最近注册的作者。

图 2-9

请求获取/作者

接下来，我们将创建一个端点，使用作者 ID 返回作者，然后更新端点，使用作者 ID 更新作者详细信息，最后一个端点使用作者 ID 删除作者。

对于通过 ID 获取作者，我们将有一个类似/authors/ 的路径，它将从请求参数中获取作者 ID 并找到匹配的作者。

将以下代码添加到 GET author by ID 端点的 GET all authors 路径下。

@app.route('/authors/<id>', methods = ['GET'])
def get_author_by_id(id):get_author = Authors.objects.get_or_404(id=ObjectId(id))author_schema = AuthorsSchema(only=['id', 'name', 'specialisation'])author, error = author_schema.dump(get_author)return make_response(jsonify({"author": author}))

现在，当您请求端点/作者/ 时，它将返回用户匹配的 ObjectId(图 2-10 )。

图 2-10

按 ID 获取作者

因此，接下来我们将创建 PUT 端点来使用作者 ID 更新作者信息。为 PUT author 端点添加以下代码。

@app.route('/authors/<id>', methods = ['PUT'])
def update_author_by_id(id):data = request.get_json()get_author = Authors.objects.get(id=ObjectId(id))if data.get('specialisation'):get_author.specialisation = data['specialisation']if data.get('name'):get_author.name = data['name']get_author.save()get_author.reload()author_schema = AuthorsSchema(only=['id', 'name', 'specialisation'])author, error = author_schema.dump(get_author)return make_response(jsonify({"author": author}))

打开 Postman 并点击与我们在其他模块中所做的相同的路由来更新作者信息，但是这里使用 GET 端点中返回的 ObjectID。

图 2-11

放置作者端点

正如您在图 2-11 中看到的，我们能够使用 PUT 端点更新作者专业化。接下来，我们将创建删除端点，使用作者 ID 删除作者，以完成我们的 CRUD 应用。

添加以下代码，为我们的应用创建删除端点。

@app.route('/authors/<id>', methods = ['DELETE'])
def delete_author_by_id(id):Authors.objects(id=ObjectId(id)).delete()return make_response("",204)

现在，让我们使用作者 ID 删除新创建的作者，与上一个应用类似，这个端点不会返回任何数据，只会返回 204 状态代码。

使用您之前使用的作者 ID 请求删除端点，它将返回如图 2-12 所示的类似响应。

图 2-12

删除作者端点

这就结束了我们的 flask-mongo CRUD 应用，app.py 中的最终代码应该是这样的。

from flask import Flask, request, jsonify, make_response
from flask_mongoengine import MongoEngine
from marshmallow import Schema, fields, post_load
from bson import ObjectIdapp = Flask(__name__)
app.config['MONGODB_DB'] = 'DB_NAME'
db = MongoEngine(app)Schema.TYPE_MAPPING[ObjectId] = fields.Stringclass Authors(db.Document):name = db.StringField()specialisation = db.StringField()class AuthorsSchema(Schema):name = fields.String(required=True)specialisation = fields.String(required=True)@app.route('/authors', methods = ['GET'])
def index():get_authors = Authors.objects.all()author_schema = AuthorsSchema(many=True, only=['id', 'name', 'specialisation'])authors, error = author_schema.dump(get_authors)return make_response(jsonify({"authors": authors}))@app.route('/authors/<id>', methods = ['GET'])
def get_author_by_id(id):get_author = Authors.objects.get_or_404(id=ObjectId(id))author_schema = AuthorsSchema(only=['id', 'name', 'specialisation'])author, error = author_schema.dump(get_author)return make_response(jsonify({"author": author}))@app.route('/authors/<id>', methods = ['PUT'])
def update_author_by_id(id):data = request.get_json()get_author = Authors.objects.get(id=ObjectId(id))if data.get('specialisation'):get_author.specialisation = data['specialisation']if data.get('name'):get_author.name = data['name']get_author.save()get_author.reload()author_schema = AuthorsSchema(only=['id', 'name', 'specialisation'])author, error = author_schema.dump(get_author)return make_response(jsonify({"author": author}))@app.route('/authors/<id>', methods = ['DELETE'])
def delete_author_by_id(id):Authors.objects(id=ObjectId(id)).delete()return make_response("",204)@app.route('/authors', methods = ['POST'])
def create_author():data = request.get_json()author = Authors(name=data['name'],specialisation=data['specialisation'])author.save()author_schema = AuthorsSchema(only=['id','name', 'specialisation'])authors, error = author_schema.dump(author)return make_response(jsonify({"author": authors}),201)if __name__ == "__main__":app.run(debug=True)

结论

现在我们已经介绍了 SQLAlchemy 和 MongoEngine，并使用它们创建了示例 CRUD 应用。在下一章，我们将详细讨论 REST API 的架构，并为我们的 Flask REST API 应用建立基础。

三、Flask CURD 应用：第一部分

在上一章中，我们讨论了数据库并实现了基于 NoSQL 和 SQL 的例子。在本章中，我们将从头开始创建一个 RESTful Flask 应用。这里我们将维护一个作者对象的数据库，以及他们写的书。这个应用将有一个用户认证机制，只允许登录的用户执行某些功能。我们现在将为 REST 应用创建以下 API 端点:

1. GET /authors:这将获取作者及其书籍的列表。

2. GET /authors/ :获取带有指定 ID 的作者及其书籍。

3. POST /authors:这将创建一个新的 Author 对象。

4. PUT /authors/ :这将编辑具有给定 ID 的作者对象。

5. DELETE /authors/ :这将删除具有给定 ID 的作者。

6. GET /books:这将返回所有的书。

7. GET /books/ :获取指定 ID 的书籍。

8. POST /books:这将创建一个新的 book 对象。

9. PUT / books/ :这将编辑给定 ID 的 book 对象。

10. DELETE /book/ :删除给定 ID 的图书。

让我们直接进入主题，我们将从创建一个新项目开始，并将其命名为 author-manager。因此，创建一个新目录，并从创建一个新的虚拟环境开始。

$ mkdir author-manager && cd author-manager$ virtualenv venv

现在我们应该有我们的虚拟环境设置；接下来，我们需要激活环境并安装依赖项，就像我们在上一章中所做的那样。

我们将从安装以下依赖项开始，并在需要时添加更多依赖项。

(venv) $ pip install flask flask-sqlalchemy marshmallow-sqlalchemy

我们还将在这个应用中使用蓝图。Flask 使用蓝图的概念来制作应用组件，并支持应用中的通用模式。蓝图有助于为应用创建更小的模块，使其易于管理。Blueprint 对于大型应用非常有价值，它简化了大型应用的工作方式。

我们将应用构建成小模块，并将所有应用代码保存在 app 文件夹内的/src 文件夹中。因此，在当前工作目录下创建一个 src 文件夹，然后在其中创建 run.py 文件。

(venv) $ mkdir src && cd src

在 src 文件夹中，我们将有我们的 run.py 文件和另一个名为 api 的目录，它将导出我们的模块，所以继续在 src 中创建一个 api 文件夹。我们将在 src 内的 main.py 文件中初始化我们的 Flask 应用，然后创建另一个文件 run.py，该文件将导入 main.py、config 文件并运行应用。

先说 main.py。

添加以下代码以导入所需的库，然后初始化 app 对象。这里我们将定义一个函数，它将接受应用配置，然后初始化我们的应用。

import os
from flask import Flask
from flask import jsonifyapp = Flask(__name__)if os.environ.get('WORK_ENV') == 'PROD':app_config = ProductionConfig
elif os.environ.get('WORK_ENV') == 'TEST':app_config = TestingConfig
else:app_config = DevelopmentConfigapp.config.from_object(app_config)if __name__ == "__main__":app.run(port=5000, host="0.0.0.0", use_reloader=False)

这就是我们 main.py 的框架。接下来，我们将创建 run.py 来调用 app 并运行应用。稍后我们将添加路由，初始化我们的 db 对象，并在 main.py 中配置日志记录。

将以下代码添加到 run.py 中，以导入 create_app 并运行应用。

from main import app as applicationif __name__ == "__main__":application.run()

这里我们已经定义了配置，导入了 create_app，并初始化了应用。接下来，我们将把配置移动到一个单独的目录，并指定特定于环境的配置。我们将在 src 中创建另一个目录/api，并从 api 目录中导出配置、模型和路由，因此现在在 src 中创建一个名为 api 的目录，然后在 api 中创建另一个名为 config 的目录。

注意

创建一个名为 init 的空文件。py，让 Python 知道它包含模块。

现在，在 config 目录中创建 config.py 和 init.py

class Config(object):DEBUG = FalseTESTING = FalseSQLALCHEMY_TRACK_MODIFICATIONS = Falseclass ProductionConfig(Config):SQLALCHEMY_DATABASE_URI =  <Production DB URL>class DevelopmentConfig(Config):DEBUG = TrueSQLALCHEMY_DATABASE_URI =  <Development DB URL>SQLALCHEMY_ECHO = Falseclass TestingConfig(Config):TESTING = TrueSQLALCHEMY_DATABASE_URI = <Testing DB URL>SQLALCHEMY_ECHO = False

前面的代码定义了我们在 main.py 中所做的基本配置，然后在顶部添加了特定于环境的配置。

因此，除了 main 之外，我们还从配置模块导入开发、测试和生产配置，并导入 OS 模块以读取环境模块。之后，我们检查是否提供了 WORK_ENV 环境变量来相应地启动应用；否则，我们默认使用开发配置启动应用。

所以我们已经提供了数据库配置，但还没有在我们的应用中初始化数据库；接下来，我们现在就开始吧。

现在在 api 中创建另一个名为 utils 的目录，它将保存我们的实用程序模块；现在，我们将在那里启动我们的 db 对象。

在实用程序中创建 database.py，并在其中添加以下代码。

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

这将开始创建我们的数据库对象；接下来，我们将在 main.py 中导入 db 对象并初始化它。

在我们导入库的地方添加下面的代码，以导入 db 对象。

from api.utils.database import dbdef create_app(config):app = Flask(__name__)app.config.from_object(config)db.init_app(app)with app.app_context():db.create_all()return app

并更新 create_app 来初始化 db 对象。

现在我们有了 REST 应用的基础，您的应用结构应该是这样的。

venv/
src
├── api/
│   ├── __init__.py
│   ├── utils
│   │     └── __init__.py
│   │     └── database.py
│   └── config
│           └── __init__.py
│           └── database.py
├── run.py
├── main.py
└── requirements.txt

接下来让我们定义我们的数据库模式。这里我们将处理两个资源，即作者和书。所以让我们先创建图书模式。我们将把所有的模式放在 api 目录中一个名为 models 的目录中，所以继续启动 models 模块，然后创建 books.py

将以下代码添加到 books.py 中，以创建 books 模型。

from api.utils.database import db
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fieldsclass Book(db.Model):__tablename__ = 'books'id = db.Column(db.Integer, primary_key=True, autoincrement=True)title = db.Column(db.String(50))year = db.Column(db.Integer)author_id = db.Column(db.Integer, db.ForeignKey('authors.id'))def __init__(self, title, year, author_id=None):self.title = titleself.year = yearself.author_id = author_iddef create(self):db.session.add(self)db.session.commit()return selfclass BookSchema(ModelSchema):class Meta(ModelSchema.Meta):model = Booksqla_session = db.sessionid = fields.Number(dump_only=True)title = fields.String(required=True)year = fields.Integer(required=True)author_id = fields.Integer()

这里我们正在导入 db 模块 marshmallow，就像我们之前做的那样来映射字段并帮助我们返回 JSON 对象。

注意，这里有一个字段 author_id，它是 authors 模型中 id 字段的外键。接下来，我们将创建 authors.py 和创建 authors 模型。

from api.utils.database import db
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields
from api.models.books import BookSchemaclass Author(db.Model):__tablename__ = 'authors'id = db.Column(db.Integer, primary_key=True, autoincrement=True)first_name = db.Column(db.String(20))last_name = db.Column(db.String(20))created = db.Column(db.DateTime, server_default=db.func.now())books = db.relationship('Book', backref="Author", cascade="all, delete-orphan")def __init__(self, first_name, last_name, books=[]):self.first_name = first_nameself.last_name = last_nameself.books = booksdef create(self):db.session.add(self)db.session.commit()return selfclass AuthorSchema(ModelSchema):class Meta(ModelSchema.Meta):model = Authorsqla_session = db.sessionid = fields.Number(dump_only=True)first_name = fields.String(required=True)last_name = fields.String(required=True)created = fields.String(dump_only=True)books = fields.Nested(BookSchema, many=True, only=['title','year','id']

前面的代码将创建我们的作者模型。请注意，我们还在这里导入了 books 模型，并创建了作者和他们的书籍之间的关系，这样当我们检索 author 对象时，我们还可以获得与其 ID 相关联的书籍，因此我们在这个模型中建立了作者和书籍之间的一对多关系。

现在，一旦我们有了 DB 模式，接下来我们需要开始创建我们的路由，但是在我们开始编写路由之前，作为应用模块化的一部分，我们还应该做一件事，创建另一个模块 responses.py 来创建 HTTP 响应的标准类。

之后，我们将在 main.py 中创建全局 HTTP 配置

在 api/utils 内部创建 responses.py，这里我们将使用来自 Flask 库的 jsonify 和 make_response 为我们的 api 创建标准响应。

因此，在 responses.py 中编写以下代码来启动该模块。

from flask import make_response, jsonifydef response_with(response, value=None, message=None, error=None, headers={}, pagination=None):result = {}if value is not None:result.update(value)if response.get('message', None) is not None:result.update({'message': response['message']})result.update({'code': response['code']})if error is not None:result.update({'errors': error})if pagination is not None:result.update({'pagination': pagination})headers.update({'Access-Control-Allow-Origin': '*'})headers.update({'server': 'Flask REST API'})return make_response(jsonify(result), response['http_code'], headers)

前面的代码公开了一个函数 response_with，供我们的 API 端点使用和响应；除此之外，我们还将创建标准响应代码和消息。

这里是我们的应用将支持的响应列表。

表 3-1 提供了我们将在应用中使用的 HTTP 响应。在 response_with 上面添加以下代码，以便在 responses.py 中定义它们。

表 3-1

HTTP 响应

| Two hundred | 200 好吧 | 对 HTTP 请求的标准响应 | | Two hundred and one | 201 已创建 | 意味着请求得到满足，并且创建了新的资源 | | Two hundred and four | 204 无内容 | 请求成功，但未返回任何数据 | | four hundred | 400 个错误请求 | 意味着由于客户端错误，服务器无法处理请求 | | Four hundred and three | 403 未授权 | 有效的请求，但发出请求的客户端无权获取资源 | | Four hundred and four | 404 未找到 | 服务器上不存在请求的资源 | | Four hundred and twenty-two | 422 无法处理的实体 | 由于语义错误，无法处理请求 | | Five hundred | 500 内部服务器错误 | 暗示服务器中出现意外情况的一般错误 |

INVALID_FIELD_NAME_SENT_422 = {"http_code": 422,"code": "invalidField","message": "Invalid fields found"}INVALID_INPUT_422 = {"http_code": 422,"code": "invalidInput","message": "Invalid input"
}MISSING_PARAMETERS_422 = {"http_code": 422,"code": "missingParameter","message": "Missing parameters."
}BAD_REQUEST_400 = {"http_code": 400,"code": "badRequest","message": "Bad request"
}SERVER_ERROR_500 = {"http_code": 500,"code": "serverError","message": "Server error"
}SERVER_ERROR_404 = {"http_code": 404,"code": "notFound","message": "Resource not found"
}UNAUTHORIZED_403 = {"http_code": 403,"code": "notAuthorized","message": "You are not authorised to execute this."
}SUCCESS_200 = {'http_code': 200,'code': 'success'
}SUCCESS_201 = {'http_code': 201,'code': 'success'
}SUCCESS_204 = {'http_code': 204,'code': 'success'
}

现在我们应该有我们的工作响应。py 模块；接下来，我们将添加用于处理错误的全局 HTTP 配置。

接下来在 main.py 中导入 status 和 response_with 函数。

from api.utils.responses import response_with
import api.utils.responses as resp

然后在 db.init_app 函数上方添加以下代码来配置全局 HTTP 配置。

    @app.after_requestdef add_header(response):return response@app.errorhandler(400)def bad_request(e):logging.error(e)return response_with(resp.BAD_REQUEST_400)@app.errorhandler(500)def server_error(e):logging.error(e)return response_with(resp.SERVER_ERROR_500)@app.errorhandler(404)def not_found(e):logging.error(e)return response_with(resp. SERVER_ERROR_404)

下面的代码添加了错误情况下的全局响应。现在你的 main.py 应该是这样的。

from flask import Flask
from flask import jsonify
from api.utils.database import db
from api.utils.responses import response_with
import api.utils.responses as respapp = Flask(__name__)if os.environ.get('WORK_ENV') == 'PROD':app_config = ProductionConfig
elif os.environ.get('WORK_ENV') == 'TEST':app_config = TestingConfig
else:app_config = DevelopmentConfigapp.config.from_object(app_config)db.init_app(app)
with app.app_context():db.create_all()# START GLOBAL HTTP CONFIGURATIONS
@app.after_request
def add_header(response):return response@app.errorhandler(400)
def bad_request(e):logging.error(e)return response_with(resp.BAD_REQUEST_400)@app.errorhandler(500)
def server_error(e):logging.error(e)return response_with(resp.SERVER_ERROR_500)@app.errorhandler(404)
def not_found(e):logging.error(e)return response_with(resp.SERVER_ERROR_404)db.init_app(app)
with app.app_context():db.create_all()if __name__ == "__main__":app.run(port=5000, host="0.0.0.0", use_reloader=False)

接下来，我们需要创建 API 端点，并使用蓝图将它们包含在 main.py 中。

我们将把我们的路由放在 api 中一个名为 routes 的目录中，所以继续创建这个文件夹；接下来添加 authors.py 来创建图书路线。

接下来，使用下面的代码导入所需的模块。

from flask import Blueprint
from flask import request
from api.utils.responses import response_with
from api.utils import responses as resp
from api.models.authors import Author, AuthorSchema
from api.utils.database import db

这里，我们从 Flask 导入 Blueprint 和 request 模块，从 responses util、Author schema 和 db 对象导入 resp _ with 和 resp 方法。

接下来，我们将配置蓝图。

author_routes = Blueprint("author_routes", __name__)

一旦完成，我们可以从我们的 POST author 路径开始，并在 book_routes 下面添加以下代码。

@author_routes.route('/', methods=['POST'])
def create_author():try:data = request.get_json()author_schema = AuthorSchema()author, error = author_schema.load(data)result = author_schema.dump(author.create()).datareturn response_with(resp.SUCCESS_201, value={"author": result})except Exception as e:print ereturn response_with(resp.INVALID_INPUT_422)

因此，前面的代码将从请求中获取 JSON 数据，并在 Author 模式上执行 create 方法，然后使用 response_with 方法返回响应，为该端点提供响应类型 201 和数据值，该数据值是带有新创建作者的 JSON 对象。

现在，在我们设置所有其他路线之前，让我们在应用中注册 author routes Blueprint，并运行应用来测试一切是否正常。

所以在你的 main.py 中，导入作者路线，然后注册蓝图。

from api.routes.authors import author_routes

然后在@app.after_request 的正上方添加下面一行。

app.register_blueprint(author_routes, url_prefix='/api/authors')

现在使用 Python run.py 命令运行应用，我们的 Flask 服务器应该启动并运行了。

让我们试试 POST authors 端点，因此在http://localhost:5000/api/authors/用下面的 JSON 数据打开 postmand 请求。

{"first_name" : "kunal","last_name" : "Relan"}

图 3-1

帖子作者端点

如您所见，books 是一个空数组，因为我们还没有创建任何书；接下来让我们添加 GET authors 端点(图 3-1 )。

@author_routes.route('/', methods=['GET'])
def get_author_list():fetched = Author.query.all()author_schema = AuthorSchema(many=True, only=['first_name', 'last_name','id'])authors, error = author_schema.dump(fetched)return response_with(resp.SUCCESS_200, value={"authors": authors})

前面的代码将添加 GET all authors route，这里我们将用一个只包含作者 ID、名字和姓氏的作者数组来响应。所以让我们来测试一下。

图 3-2

获取作者路线

正如您在图 3-2 中看到的，端点响应了一组作者。

接下来，让我们添加另一个 GET route 来使用作者的 ID 获取特定的作者，并添加以下代码来添加该 route。

@author_routes.route('/<int:author_id>', methods=['GET'])
def get_author_detail(author_id):fetched = Author.query.get_or_404(author_id)author_schema = AuthorSchema()author, error = author_schema.dump(fetched)return response_with(resp.SUCCESS_200, value={"author": author})

前面的代码从 route 参数中获取一个整数，查找具有相应 ID 的作者，并返回 author 对象。

因此，让我们尝试获取 ID 为 1 的作者(图 3-3 )。

图 3-3

正在获取 ID 为 1 的作者

如果具有该 ID 的作者存在，我们将得到带有 200 状态代码和作者对象的响应，否则为 404，如下图所示。正如您所看到的，没有 ID 为 2 的作者，get_or_404 方法在端点上抛出 404 错误，然后由 app.errorhandler(404)按照我们在 main.py 中提到的方式进行处理(图 3-4 )。

图 3-4

找不到 ID 为 2 的作者

在我们继续为 author 对象创建 PUT 和 DELETE 端点之前，让我们启动图书路由。在同一个 routes 文件夹中创建 books.py，并添加以下代码来启动路由。

from flask import Blueprint, request
from api.utils.responses import response_with
from api.utils import responses as resp
from api.models.books import Book, BookSchema
from api.utils.database import dbbook_routes = Blueprint("book_routes", __name__)

然后在 main.py 中注册 book routes，就像我们注册 author routes 一样。将下面的代码添加到导入作者路径的位置。

from api.routes.books import book_routes

然后，在您添加了 author route blueprint 注册的地方的正下方，添加以下代码。

app.register_blueprint(book_routes, url_prefix='/api/books')

现在您的 main.py 应该有以下代码。

import logging
import sys
import api.utils.responses as resp
from flask import Flask, jsonify
from api.utils.database import db
from api.utils.responses import response_with
from api.routes.authors import author_routes
from api.routes.books import book_routesdef create_app(config):app = Flask(__name__)app.config.from_object(config)db.init_app(app)with app.app_context():db.create_all()app.register_blueprint(author_routes, url_prefix='/api/authors')app.register_blueprint(book_routes, url_prefix='/api/books')@app.after_requestdef add_header(response):return response@app.errorhandler(400)def bad_request(e):logging.error(e)return response_with(resp.BAD_REQUEST_400)@app.errorhandler(500)def server_error(e):logging.error(e)return response_with(resp.SERVER_ERROR_500)@app.errorhandler(404)def not_found(e):logging.error(e)return response_with(resp.SERVER_ERROR_404)db.init_app(app)with app.app_context():db.create_all()logging.basicConfig(stream=sys.stdout,format='%(asctime)s|%(levelname)s|%(filename)s:%(lineno)s|%(message)s',level=logging.DEBUG)return app

接下来让我们从创建 POST book 端点开始；打开 routes 文件夹中的 books.py，并在 book_routes 下添加以下代码。

@book_routes.route('/', methods=['POST'])
def create_book():try:data = request.get_json()book_schema = BookSchema()book, error = book_schema.load(data)result = book_schema.dump(book.create()).datareturn response_with(resp.SUCCESS_201, value={"book": result})except Exception as e:print ereturn response_with(resp.INVALID_INPUT_422)

前面的代码将获取用户数据，然后对 book schema 执行 create()方法，就像我们在 author object 中所做的一样；让我们保存文件并测试端点。

{"title" : "iOS Penetration Testing","year" : 2016,"author_id": 1}

我们将使用前面的 JSON 数据发送到端点，我们应该得到一个包含 200 状态代码和新创建的 book 对象的响应。正如我们之前所讨论的，我们已经建立了作者和书籍之间的关系，在前面的例子中，我们已经为新书指定了 ID 为 1 的作者，所以一旦这个 API 成功，我们将能够获取 ID 为 1 的作者，作为响应，books 数组将把这本书作为一个对象(图 3-5 )。

图 3-5

获取 ID 为 1 的作者

正如您在图 3-6 中看到的，当我们请求/authors/1 端点时，除了作者详细信息，我们还获得了 books 数组，其中包含作者链接到的书籍列表。

图 3-6

获取作者端点

所以我们的模特关系很好。现在我们可以继续为作者路由创建其余的端点。继续添加下面的代码来获取作者路由的 PUT 端点，以更新作者对象。

@author_routes.route('/<int:id>', methods=['PUT'])
def update_author_detail(id):data = request.get_json()get_author = Author.query.get_or_404(id)get_author.first_name = data['first_name']get_author.last_name = data['last_name']db.session.add(get_author)db.session.commit()author_schema = AuthorSchema()author, error = author_schema.dump(get_author)return response_with(resp.SUCCESS_200, value={"author": author})

前面的代码将创建我们的 PUT 端点来更新 author 对象。在前面的代码中，我们在数据变量中获取一个请求 JSON，然后用请求参数中提供的 ID 获取作者。如果没有找到具有该 ID 的作者，请求以 404 状态代码结束，或者 get_author 包含 author 对象，然后我们用请求 JSON 中提供的数据更新 first_name 和 last_name，然后保存会话。

所以让我们继续更新我们不久前创建的作者的名字和姓氏(图 3-7 )。

图 3-7

放置作者端点

所以这里我们更新了作者的名和姓。然而，在 PUT 中，我们需要发送对象的整个请求体，正如我们在第二章中所讨论的，所以接下来我们将创建一个补丁端点来只更新 author 对象的一部分。为修补程序端点添加以下代码。

@author_routes.route('/<int:id>', methods=['PATCH'])
def modify_author_detail(id):data = request.get_json()get_author = Author.query.get(id)if data.get('first_name'):get_author.first_name = data['first_name']if data.get('last_name'):get_author.last_name = data['last_name']db.session.add(get_author)db.session.commit()author_schema = AuthorSchema()author, error = author_schema.dump(get_author)return response_with(resp.SUCCESS_200, value={"author": author})

前面的代码像另一个端点一样获取请求 JSON，但并不期望整个请求体，而只是请求体中需要更新的字段，同样，它更新 author 对象并保存会话。让我们尝试一下，这次我们将只更改 author 对象的名字。

图 3-8

更改作者对象的名字

正如您在图 3-8 中看到的，我们只在请求体中提供了名字，它已经被更新了。接下来，我们将最终创建删除作者端点，它将从请求参数中获取作者 ID 并删除作者对象。注意，在这个例子中，我们将用 204 状态码来响应，没有任何内容。

@author_routes.route('/<int:id>', methods=['DELETE'])
def delete_author(id):get_author = Author.query.get_or_404(id)db.session.delete(get_author)db.session.commit()return response_with(resp.SUCCESS_204)

添加前面的代码，现在这将创建我们的删除端点。让我们继续尝试删除 ID 为 1 的作者(图 3-9 )。

图 3-9

删除作者端点

有了这个端点，我们的 author 对象应该从数据库中删除，并且在创建 author 模式时，我们配置了 book 关系中的所有级联。因此，所有与作者 ID 1 相关的书籍也将被删除，以确保我们没有任何没有作者 ID 的书籍。

这就是我们的作者路线，接下来我们将致力于我们的书端点的其余部分。接下来，在 books.py 中添加以下代码来创建 GET books 端点。

@book_routes.route('/', methods=['GET'])
def get_book_list():fetched = Book.query.all()book_schema = BookSchema(many=True, only=['author_id','title', 'year'])books, error = book_schema.dump(fetched)return response_with(resp.SUCCESS_200, value={"books": books})

保存文件并尝试端点；现在你将得到一个空数组，因为当我们删除作者时，作者 ID 为 1 的书也被删除了。

图 3-10

获取书籍端点

正如您在图 3-10 中看到的，到目前为止，表中没有书籍，所以继续创建一个作者，然后添加几本具有该作者 ID 的书籍，因为我们不能添加没有作者的书籍，否则它会以 422 不可处理的实体错误结束。

接下来，我们将通过 ID 端点创建 GET Book。

@book_routes.route('/<int:id>', methods=['GET'])
def get_book_detail(id):fetched = Book.query.get_or_404(id)book_schema = BookSchema()books, error = book_schema.dump(fetched)return response_with(resp.SUCCESS_200, value={"books": books})

以下代码将创建按 ID 端点获取图书；接下来，我们将创建 PUT、PATCH 和 DELETE 端点，并为其添加以下代码。

book_routes.route('/<int:id>', methods=['PUT'])
def update_book_detail(id):data = request.get_json()get_book = Book.query.get_or_404(id)get_book.title = data['title']get_book.year = data['year']db.session.add(get_book)db.session.commit()book_schema = BookSchema()book, error = book_schema.dump(get_book)return response_with(resp.SUCCESS_200, value={"book": book})@book_routes.route('/<int:id>', methods=['PATCH'])
def modify_book_detail(id):data = request.get_json()get_book = Book.query.get_or_404(id)if data.get('title'):get_book.title = data['title']if data.get('year'):get_book.year = data['year']db.session.add(get_book)db.session.commit()book_schema = BookSchema()book, error = book_schema.dump(get_book)return response_with(resp.SUCCESS_200, value={"book": book})@book_routes.route('/<int:id>', methods=['DELETE'])
def delete_book(id):get_book = Book.query.get_or_404(id)db.session.delete(get_book)db.session.commit()return response_with(resp.SUCCESS_204)

因此，这将结束我们的图书和作者路线，现在我们有一个工作休息应用。现在，您可以尝试在 author 和 book 路径上执行 CRUD。

用户认证

一旦我们准备好了所有的路由，我们需要添加用户身份验证以确保只有登录的用户才能访问特定的路由，所以现在我们将添加用户登录和注册路由，但在此之前，我们需要添加用户模式。

在模型中创建 users.py。在模式中，我们将添加两个静态方法来加密密码和验证密码，为此我们需要一个名为 passlib 的 Python 库，所以在创建模式之前，让我们使用 PIP 安装 passlib。

(venv)$ pip install passlib

完成后，添加以下代码来添加用户模式和方法。

from api.utils.database import db
from passlib.hash import pbkdf2_sha256 as sha256
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fieldsclass User(db.Model):__tablename__ = 'users'id = db.Column(db.Integer, primary_key = True)username = db.Column(db.String(120), unique = True, nullable = False)password = db.Column(db.String(120), nullable = False)def create(self):db.session.add(self)db.session.commit()return self@classmethoddef find_by_username(cls, username):return cls.query.filter_by(username = username).first()@staticmethoddef generate_hash(password):return sha256.hash(password)@staticmethoddef verify_hash(password, hash):return sha256.verify(password, hash)class UserSchema(ModelSchema):class Meta(ModelSchema.Meta):model = Usersqla_session = db.sessionid = fields.Number(dump_only=True)username = fields.String(required=True)

因此，我们在这里添加了一个类方法，通过用户名查找用户，创建一个用户，然后创建两个静态方法来生成散列并验证它。我们将在创建用户路线时使用这些方法。

接下来在 routes 目录中创建 users.py，这是我们添加用户登录和注册路由的地方。

对于跨应用的用户认证，我们将使用 JWT (JSON Web 令牌)认证。JWT 是一个开放标准，它定义了一种紧凑的、自包含的方式，以 JSON 对象的形式安全地传输信息。JWT 是 REST 世界中一种流行的用户授权方式。在 Flask 中有一个名为 Flask-JWT 扩展的开源扩展，它提供了 JWT 支持和其他有用的方法。

让我们继续安装 Flask-JWT-扩展。

(venv)$ pip install flask-jwt-extended

接下来，我们将在 main.py 中初始化应用中的 JWT 模块，以便在 main.py 中导入库

from flask_jwt_extended import JWTManager

接下来用 db.init_app()上面的代码初始化 JWTManager。

jwt = JWTManager(app)

安装和初始化之后，让我们导入用户路由文件所需的模块。

from flask import Blueprint, request
from api.utils.responses import response_with
from api.utils import responses as resp
from api.models.users import User, UserSchema
from api.utils.database import db
from flask_jwt_extended import create_access_token

这些是我们在用户路线中需要的模块；接下来，我们将使用 Blueprint 配置路由，代码如下。

user_routes = Blueprint("user_routes", __name__)

接下来，我们将在 main.py 文件中导入并注册/users 路由，因此在 main.py 中添加以下代码来导入用户路由。

from api.routes.users import user_routes

现在，在我们已经声明了其他路由的地方的正下方，添加以下代码行。

app.register_blueprint(user_routes, url_prefix='/api/users')

接下来，我们将创建我们的 POST 用户路由来创建一个新用户，并在 routes 内部的 users.py 中添加以下代码。

@user_routes.route('/', methods=['POST'])
def create_user():try:data = request.get_json()data['password'] = User.generate_hash(data['password'])user_schmea = UserSchema()user, error = user_schmea.load(data)result = user_schmea.dump(user.create()).datareturn response_with(resp.SUCCESS_201)except Exception as e:print ereturn response_with(resp.INVALID_INPUT_422)

这里，我们将用户请求数据放在一个变量中，然后对密码执行 generate_hash()函数并创建用户。一旦完成，我们将返回一个 201 响应。

接下来，我们将为注册用户创建一个登录路径。为相同的添加以下代码。

@user_routes.route('/login', methods=['POST'])
def authenticate_user():try:data = request.get_json()current_user = User.find_by_username(data['username'])if not current_user:return response_with(resp.SERVER_ERROR_404)if User.verify_hash(data['password'], current_user.password):access_token = create_access_token(identity = data['username'])return response_with(resp.SUCCESS_201, value={'message': 'Logged in as {}'.format(current_user.username), "access_token": access_token})else:return response_with(resp.UNAUTHORIZED_401)except Exception as e:print ereturn response_with(resp.INVALID_INPUT_422)

下面的代码将从请求数据中获取用户名和密码，并使用我们在模式中创建的 find_by_username()方法检查具有所提供用户名的用户是否存在。接下来，如果用户不存在，我们将使用 404 进行响应，或者使用模式中的 verify_hash()函数来验证密码。如果用户存在，我们将生成一个 JWT 令牌并用 200 来响应；否则以 401 回应。现在我们已经有了用户登录。接下来，我们需要将 jwt required decorator 添加到我们想要保护的路由中。因此，在 routes 中导航到 authors.py，并使用下面的代码导入装饰器。

from flask_jwt_extended import jwt_required

然后在端点定义之前，使用下面的代码添加装饰器。

@jwt_required

我们将把装饰器添加到 authors.py 和 books.py 的 DELETE、PUT、POST 和 PATCH 端点，这些函数现在应该是这样的。

@author_routes.route('/', methods=['POST'])
@jwt_required
def create_author():....Function code

让我们继续测试我们的用户端点。打开 Postman，用用户名和密码请求 POST 用户端点。我们将使用下面的样本数据。

{"username" : "admin","password" : "flask2019"}

图 3-11

用户注册端点

这样我们的新用户就创建好了(图 3-11)；接下来，我们将尝试使用相同的凭据登录并获取 JWT。

图 3-12

用户登录端点

如图 3-12 所示，我们已经使用新创建的用户成功登录。现在让我们尝试访问最近添加了 jwt_required decorator 的 POST author 路径(图 3-13 )。

图 3-13

无 JWT 令牌的帖子作者路线

如图 3-14 所示，我们无法再访问 POST author 路径，jwt_required decorator 返回 401 错误。现在，让我们通过在报头中提供 JWT 来尝试访问相同的路由。在 Postman 中请求的头部分，添加带有名为 Authorization 的密钥的令牌，然后在 value 中添加 Bearer <令牌>来提供 JWT 令牌，如图 3-14 所示。

图 3-14

JWT 后作者路线

如您所见，添加 JWT 令牌后，我们能够再次访问端点，这就是我们保护 REST 端点的方式。

因此，在下面的场景中，我们允许任何人登录平台，然后访问路线。然而，在实际应用中，我们还可以进行电子邮件验证和限制用户注册，同时我们还可以启用基于用户的访问控制，不同类型的用户可以访问特定的 API。

结论

本章到此结束，我们已经成功地创建了一个带有用户认证的 REST 应用。在下一章，我们将致力于记录 REST APIs，集成单元测试，以及部署我们的应用。