网站服务器租用还是自买,东莞市疾控中心官网,网站开发php工资,网站作风建设年专栏原文#xff1a;Building REST APIs with Flask 协议#xff1a;CC BY-NC-SA 4.0 一、从 Flask 开始
Flask 是一个 BSD 许可的 Python 微框架#xff0c;基于 Werkzeug 和 Jinja2。作为一个微框架并不会降低它的功能性#xff1b;Flask 是一个非常简单但高度可扩展的框架。… 原文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 APIsHeroku 是一个云平台有助于 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 服务是使用以下特性定义和实现的: 统一界面 陈述 信息 资源之间的链接 贮藏 无国籍的
统一界面
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 version1.0 encodingUTF-8?
BookID 1 /IDName Building REST APIs with Flask /NameAuthor Kunal Relan /AuthorPublisher Apress / Publisher
/BookListing 1-2XML Representation of a Book Resource
信息
在 REST 架构中消息是一个重要的关键它本质上建立了客户机-服务器风格的数据通信方式。客户端和服务器通过消息相互通信其中客户端向服务器发送消息这通常被称为请求而服务器发送响应。除了客户端和服务器之间以请求和响应主体的形式交换的实际数据之外客户端和服务器还以请求和响应头的形式交换一些元数据。HTTP 1.1 以如下方式定义了请求和响应头格式以便在不同种类的系统之间实现统一的数据通信方式(图 1-3 )。 图 1-3
HTTP 示例请求
在图 1-4 中GET 是请求方法“/comments”是服务器中的路径“postId1”是请求参数“HTTP/1.1”是客户端正在请求的协议版本“jsonplaceholder.typicode.com”是服务器主机内容类型是请求头的一部分。所有这些结合起来就是服务器能够理解的 HTTP 请求。
作为回报HTTP 服务器发送对所请求资源的响应。
[{postId: 1,id: 1,name: id labore ex et quam laborum,email: Eliseogardner.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_Kuhicsydney.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: Nikitagarfield.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: Lewalysha.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: Haydenalthea.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 时需要检查的事项列表: 理解用例。知道你为什么要构建 API 以及 API 将提供什么服务是非常重要的。 列出 API 特性来理解你的 API 将要做的所有动作。这还包括列出行动并将它们组合在一起以处理冗余的端点。 确定将使用 API 的不同平台并相应地提供支持。 对支持增长和扩展基础架构进行长期规划。 规划 API 版本控制策略确保对不同版本的 API 提供持续支持。 规划 API 访问策略即身份验证、ACL 和限制。 计划 API 文档和测试。 理解如何在你的 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 下载的 PythonPIP 会预装 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
虚拟环境目录结构
下面是结构中每个文件夹包含的内容: bin:与虚拟环境交互的文件。 include: C 头文件来编译 Python 包。 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_APPapp.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 时以下发行版会随主框架一起安装: Werkzeug(http://werkzeug.pocoo.org/*):*Werkzeug 实现了 WSGI应用和服务器之间的标准 Python 接口。 Jinja ( http://jinja.pocoo.org/ ): Jinja 是 Flask 中的模板引擎为应用呈现页面。 MarkupSafe ( its dangerous(https://pythonhosted.org/itsdangerous/):its dangerous 负责安全地对数据进行签名以确保数据的完整性并用于保护 Flask 会话 cookies。 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 数据库以及它们之间的连接和交互。
在本章中我们将讨论以下主题: NoSQL 与 SQL 数据库 连接 Flask-SQLAlchemy 使用 Flask-SQLAlchemy 与 MySQL 数据库交互 连接 Flask-MongoEngine 使用 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 扩展允许本地连接到 MySQLPyMongo 是在 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] mysqlpymysql://mysql_username:mysql_passwordmysql_host:mysql_port/mysql_db
db SQLAlchemy(app)if __name__ __main__:app.run(debugTrue)
这里我们导入 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 CTRLC 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_keyTrue)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_onlyTrue)name fields.String(requiredTrue)specialisation fields.String(requiredTrue)
前面的代码将变量属性映射到字段对象在 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(manyTrue)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_keyTrue)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] mysqlpymysql://mysql_username:mysql_passwordmysql_host:mysql_port/mysql_dbdb SQLAlchemy(app)class Authors(db.Model):id db.Column(db.Integer, primary_keyTrue)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_onlyTrue)name fields.String(requiredTrue)specialisation fields.String(requiredTrue)app.route(/authors, methods [GET])
def index():get_authors Authors.query.all()author_schema AuthorsSchema(manyTrue)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(debugTrue)
因此我们现在已经创建并测试了我们的样本 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(debugTrue)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(requiredTrue)specialisation fields.String(requiredTrue)
前面的代码让我们创建一个模式我们将使用该模式将 db 对象映射到 marshmallow。请注意这里我们没有使用 marshmallow-sqlalchemy它对 sqlalchemy 有一个额外的支持层因此代码看起来略有变化。
现在我们可以编写 GET 端点来从数据库中获取所有作者。
app.route(/authors, methods [GET])
def index():get_authors Authors.objects.all()author_schema AuthorsSchema(manyTrue,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 CTRLC 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(namedata[name],specialisationdata[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(idObjectId(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(idObjectId(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(idObjectId(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(requiredTrue)specialisation fields.String(requiredTrue)app.route(/authors, methods [GET])
def index():get_authors Authors.objects.all()author_schema AuthorsSchema(manyTrue, 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(idObjectId(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(idObjectId(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(idObjectId(id)).delete()return make_response(,204)app.route(/authors, methods [POST])
def create_author():data request.get_json()author Authors(namedata[name],specialisationdata[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(debugTrue)
结论
现在我们已经介绍了 SQLAlchemy 和 MongoEngine并使用它们创建了示例 CRUD 应用。在下一章我们将详细讨论 REST API 的架构并为我们的 Flask REST API 应用建立基础。
三、Flask CURD 应用第一部分
在上一章中我们讨论了数据库并实现了基于 NoSQL 和 SQL 的例子。在本章中我们将从头开始创建一个 RESTful Flask 应用。这里我们将维护一个作者对象的数据库以及他们写的书。这个应用将有一个用户认证机制只允许登录的用户执行某些功能。我们现在将为 REST 应用创建以下 API 端点: GET /authors:这将获取作者及其书籍的列表。 GET /authors/ :获取带有指定 ID 的作者及其书籍。 POST /authors:这将创建一个新的 Author 对象。 PUT /authors/ :这将编辑具有给定 ID 的作者对象。 DELETE /authors/ :这将删除具有给定 ID 的作者。 GET /books:这将返回所有的书。 GET /books/ :获取指定 ID 的书籍。 POST /books:这将创建一个新的 book 对象。 PUT / books/ :这将编辑给定 ID 的 book 对象。 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(port5000, host0.0.0.0, use_reloaderFalse)
这就是我们 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 URLclass DevelopmentConfig(Config):DEBUG TrueSQLALCHEMY_DATABASE_URI Development DB URLSQLALCHEMY_ECHO Falseclass TestingConfig(Config):TESTING TrueSQLALCHEMY_DATABASE_URI Testing DB URLSQLALCHEMY_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__ booksid db.Column(db.Integer, primary_keyTrue, autoincrementTrue)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_idNone):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_onlyTrue)title fields.String(requiredTrue)year fields.Integer(requiredTrue)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__ authorsid db.Column(db.Integer, primary_keyTrue, autoincrementTrue)first_name db.Column(db.String(20))last_name db.Column(db.String(20))created db.Column(db.DateTime, server_defaultdb.func.now())books db.relationship(Book, backrefAuthor, cascadeall, 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_onlyTrue)first_name fields.String(requiredTrue)last_name fields.String(requiredTrue)created fields.String(dump_onlyTrue)books fields.Nested(BookSchema, manyTrue, 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, valueNone, messageNone, errorNone, headers{}, paginationNone):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 responseapp.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 responseapp.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(port5000, host0.0.0.0, use_reloaderFalse)
接下来我们需要创建 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(manyTrue, 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 responseapp.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(streamsys.stdout,format%(asctime)s|%(levelname)s|%(filename)s:%(lineno)s|%(message)s,levellogging.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(manyTrue, 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__ usersid 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 selfclassmethoddef 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_onlyTrue)username fields.String(requiredTrue)
因此我们在这里添加了一个类方法通过用户名查找用户创建一个用户然后创建两个静态方法来生成散列并验证它。我们将在创建用户路线时使用这些方法。
接下来在 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集成单元测试以及部署我们的应用。
