Python Falcon 简明教程

Python Falcon - Quick Guide

Python Falcon - Introduction

Falcon 是一个 Python 库,用于开发任务关键型 REST API 和微服务。它同时支持 WSGI 和 ASGI 规范。Falcon 框架由 Kurt Griffiths 于 2013 年 1 月开发。Falcon 的最新版本是 3.1.0,于 2022 年 3 月发布。

Falcon 是一款轻量级的 Web 开发框架。其极简主义设计允许开发人员根据需要选择最佳策略和第三方包。

Falcon - Important Features

Falcon 根据 Apache 2.0 License 条款发布。

Falcon 的一些重要功能包括 −

  1. Falcon 的最新版本支持 ASGI、WSGI 和 WebSocket。

  2. Falcon 提供对 asyncio 的本机支持。

  3. 它的稳定接口确保了向后兼容性。

  4. Falcon 遵循 REST 架构样式构建 API。

  5. 基于类的 HTTP 资源构建。

  6. Highly-optimized, extensible code base.

  7. Falcon 通过请求和响应类轻松访问标头和正文。

  8. 可用的中间件组件和钩子,用于 DRY 请求处理。

  9. 惯用的 HTTP 错误响应和异常处理。

Falcon - Design Philosophy

Falcon 最大限度减少对象实例化数量,以免创建对象时花费太多开销,并减少内存使用。同一个实例将用于处理该路由上发来的所有请求。

  1. 异常由资源响应者(方法例如 on_get(), on_post(), 等)正确处理。Falcon 并不费力地尝试保护响应者自身代码不受攻击。高质量的 Falcon API 应满足以下要求 −

  2. Falcon 框架是线程安全的。为每个传入的 HTTP 请求创建单独的新请求和响应对象。然而,附加到路由的每个资源类的单个实例在所有请求中共享。中间件对象、钩子、以及自定义的错误处理程序,也被共享。因此,作为一个整体,你的 WSGI 应用将是线程安全的。

  3. 从版本 3.0 开始,Falcon 支持 asyncio 。使用 falcon.asgi.App 类创建异步应用程序,并通过 ASGI 应用程序服务器(例如 Uvicorn )提供服务。

  4. Falcon 的异步版本支持 ASGI WebSocket 协议。

Falcon - Comparison with Other Frameworks

Python Web 框架有两大类 − full-stackmicro 框架。

  1. 全栈框架自带内置功能和类库。 Django, Turbogears,Web2Py 是全栈框架。

  2. 相反,微框架是极简主义的,仅提供最少量的内容;因此,开发者可以自由选择官方或第三方扩展,并且只包含他们所需要的插件。 Flask, Falcon, Pyramid 属于微框架类别。

我们根据以下参数将 Falcon 框架与不同框架进行比较 −

Performance

Falcon 应用非常快,与 Flask 和 Pyramid 等微框架相比。而全栈框架通常很慢。

REST Support

Falcon 是一个旨在用于开发 REST API 和微服务的框架。FastAPI 也支持 REST 开发。Flask 和 Django 没有内置的 REST 支持。但是,可以使用扩展名启用它。

Templating

Falcon 应用程序不应提供模板网页。它未与任何模板库捆绑在一起。不过,可以使用 jinja2Macho 库。另一方面,Flask 对 jinja2 有内置支持。Django 有自己的模板库。FastAPI 还可以处理任何选择的模板库。

Database Support

在 Falcon 中,没有内置的数据库支持。可以使用 SQLAlchemy 模型与关系数据库(如 MyQL、PostgreSQL、SQLite 等)进行交互。另一方面,Django 具有自己的 ORM 框架,可以开箱即用。

Flask 应用程序还可以通过 Flask 扩展与数据库进行交互。TurboGears 的早期版本与 SQLObject ORM 库兼容。较新版本与 SQLAlchemy 兼容。

Flexibility

Falcon 应用程序非常灵活。它是要求高度定制和性能调优的应用程序的理想选择。FastAPI 和 Flask 也能灵活地进行编码,并且不会限制用户使用特定的项目或代码布局。

Security

Falcon 没有内置的安全保障支持。Django 和 FastAPI 等其他框架可确保高度的安全性。Flask 也提供了出色的防护功能,可以抵御 CSRF 和 XSS 攻击等安全威胁。

Testing

Falcon 提供了使用 unittest 和 Pytest 的内置测试支持。Flask 和 Django 也支持 unittest。FastAPI 支持 unittest 和 starlette 测试功能。

Python Falcon - Environment Setup

最新版本的 Falcon 需要 Python 3.5 或更高版本。最简单、也同时是推荐的安装方法是使用 PIP 安装程序,最好是在虚拟环境中安装。

可以通过运行以下命令安装最新稳定版本 −

pip3 install falcon

要验证是否已成功安装,请导入该库并检查其版本。

>>> import falcon
>>>falcon.__version__
'3.1.0'

要安装最新测试版,应使用以下命令 −

pip3 install --pre falcon

Falcon 从早期版本开始就支持 WSGI。Falcon 应用程序可以在 Python 的标准库模块 wsgiref 的帮助下使用内置 WSGI 服务器运行。但是,它不适用于生产环境,为此需要使用 WSGI 服务器,例如 gunicorn、waitress 或 uwsgi。

对于 Windows 上的 Falcon,可以使用 Waitress ,一个生产级的纯 Python WSGI 服务器。与往常一样,使用 pip 安装程序进行安装。

pip3 install waitress

Gunicorn 服务器无法安装在 Windows 上。但是,可以在 Windows 10 上的 Windows 子系统 Linux ( WSL ) 环境中使用它。要在 Linux、WSL 或 Docker 容器中使用 gunicorn,请使用

pip3 install gunicorn

如果你想运行异步 Falcon 应用程序,则需要一个与 ASGI 兼容的应用程序服务器。Uvicorn 服务器可在 Windows 和 Linux 系统上使用。

pip3 install uvicorn

Python Falcon - WSGI vs ASGI

Web Server Gateway Interface(WSGI)

一些最流行的 Python Web 框架实现了 WSGI(代表 Web Server Gateway Interface )。WSGI 本质上是一组规范,用于 Web 服务器和 Web 应用程序之间的通用接口,由 Web 服务器软件实现,用于处理基于 Python 的 Web 应用程序的请求。WSGI 规范最初于 2003 年引入(PEP 333),并于 2010 年进行更新(PEP 3333)。

服务器通过传递以下参数来调用 WSGI 应用程序对象 −

  1. environ − 一个 Python dict 对象,它类似于 CGI 环境变量和某些特定的 WSGI 变量。

  2. start_response − 应用程序用来返回其响应以及标头和状态代码的回调函数。

此对象可以是 Python 中的任意可调用对象,例如带 call() 方法的函数、方法、类或其实例。此应用程序对象必须返回一个包含单个字节字符串的迭代器。

def application (environ, start_response):
   ...
   ...
   return [("Hello World!".encode("utf-8")]

但是,启用 WSGI 的服务器在运行时是同步的,因此,应用程序不够高效。Python 通过引入 asyncio 模块作为标准库的一部分,从 3.4 版开始支持异步编程。

asyncio 模块提供在 Python 应用程序中并入并发编程样式的能力(通常称为协作式多任务处理)。在这种方法中,操作系统不会妨碍不同进程之间的上下文切换。取而代之的是,进程定期让步以容纳其他进程,以便许多应用程序可以同时运行。

在 Python 的 3.5 版中,添加了这两个关键字 asyncawait 。使用 async 关键字定义的 Python 函数称 coroutine 为,因此不能像普通函数一样运行。取而代之的是,我们需要使用 asyncio.run (coroutine) 调用它。可以通过 await 关键字使协程的执行暂停,直到另一个协程完成。

import asyncio
async def main():
   print('hello')
   await asyncio.sleep(5)
   print('world')

asyncio.run(main())

Asynchronous Server Gateway Interface(ASGI)

ASGI 代表 Asynchronous Server Gateway Interface (根据其官方文档,它是 WSGI 的精神继承者),它为 Python Web 服务器、应用程序和框架增加了异步功能。

ASGI 应用程序是一个异步可调用对象(用户定义的函数或具有 call() 方法的类的对象)。它获取三个参数,如下所示:

  1. Scopedict 包含特定连接的详细信息

  2. Send − 一个异步可调用对象,可通过它将事件消息发送给客户端

  3. Receive − 另一个异步可调用对象。应用程序可以从客户端接收事件消息。

下面是一个异步函数表示的简单 ASGI 应用程序的原型 −

async def app(scope, receive, send):
   assert scope['type'] == 'http'
   await send({
   'type': 'http.response.start',
   'status': 200,
   'headers': [
      [b'content-type', b'text/plain'],
   ],
})
await send({
   'type': 'http.response.body',
   'body': b'Hello, world!',
})

Python Falcon - Hello World(WSGI)

要创建一个简单的 Hello World Falcon 应用程序,首先导入库并声明 App 对象的一个实例。

import falcon
app = falcon.App()

Falcon 遵循 REST 架构风格。声明一个资源类,其中包括一个或多个表示标准 HTTP 动词的方法。以下 HelloResource 类包含 on_get() 方法,预期在服务器收到 GET 请求时调用此方法。该方法返回 Hello World 响应。

class HelloResource:
   def on_get(self, req, resp):
   """Handles GET requests"""
   resp.status = falcon.HTTP_200
   resp.content_type = falcon.MEDIA_TEXT
   resp.text = (
      'Hello World'
   )

要调用此方法,我们需要将其注册到路由或 URL。Falcon 应用程序对象通过 add_rule 方法将处理程序方法分配到对应的 URL,来处理传入请求。

hello = HelloResource()
app.add_route('/hello', hello)

Falcon 应用程序对象只不过是一个 WSGI 应用程序。我们可以使用 Python 标准库的 wsgiref 模块中的内置 WSGI 服务器。

from wsgiref.simple_server import make_server

if __name__ == '__main__':
   with make_server('', 8000, app) as httpd:
   print('Serving on port 8000...')
# Serve until process is killed
httpd.serve_forever()

Example

让我们将所有这些代码段放入 hellofalcon.py

from wsgiref.simple_server import make_server

import falcon

app = falcon.App()

class HelloResource:
   def on_get(self, req, resp):
      """Handles GET requests"""
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_TEXT
      resp.text = (
         'Hello World'
      )
hello = HelloResource()

app.add_route('/hello', hello)

if __name__ == '__main__':
   with make_server('', 8000, app) as httpd:
   print('Serving on port 8000...')
# Serve until process is killed
httpd.serve_forever()

从命令提示符运行此代码。

(falconenv) E:\falconenv>python hellofalcon.py
Serving on port 8000...

Output

在另一个终端中,按如下方式运行 Curl 命令 -

C:\Users\user>curl localhost:8000/hello
Hello World

还可以打开浏览器窗口并输入上述 URL 以获取“Hello World”响应。

hello world

Python Falcon - Waitress

不建议在生产环境中使用开发服务器。开发服务器的效率、稳定性或安全性都不高。

Waitress 是一个生产级的纯 Python WSGI 服务器,性能非常不错。除了生活在 Python 标准库中的依赖项之外,它没有其他依赖项。它在 Unix 和 Windows 上的 CPython 上运行。

确保 Waitress 服务器已安装在工作环境中。该库包含 serve 类,其对象负责处理传入请求。serve 类的构造函数需要三个参数。

serve (app, host, port)

falcon 应用程序对象是 app 参数。默认情况下,host 和 port 的默认值为 localhost 8080。listen 参数是字符串,作为 host:port 参数的组合,默认值为“0.0.0.0:8080”

Example

hellofalcon.py 代码中,我们导入 serve 类,而不是 simple_server ,并按如下方式实例化它的对象 −

from waitress import serve
import falcon
class HelloResource:
   def on_get(self, req, resp):
   """Handles GET requests"""
   resp.status = falcon.HTTP_200
   resp.content_type = falcon.MEDIA_TEXT
   resp.text = (
      'Hello World'
   )
app = falcon.App()
hello = HelloResource()
app.add_route('/hello', hello)
if __name__ == '__main__':
   serve(app, host='0.0.0.0', port=8000)

执行 hellofalcon.py 并像之前一样在浏览器中访问 http://localhost:8000/hellolink 。请注意,主机 0.0.0.0 使 localhost 公开可见。

还可以从命令行启动 Waitress 服务器,如下所示 −

waitress-serve --port=8000 hellofalcon:app

Python Falcon - ASGI

ASGI 代表 Asynchronous Server Gateway Interface (根据其官方文档,它是 WSGI 的精神继承者),它为 Python Web 服务器、应用程序和框架增加了异步功能。

要运行异步 web 应用程序,我们需要一个 ASGI 应用程序服务器。常用的选择包括 −

  1. Uvicorn

  2. Daphne

  3. Hypercorn

在本教程中,我们将在 async 示例中使用 Uvicorn 服务器。

Hello World - ASGI

Falcon 的 ASGI 相关功能在 falcon.asgi 模块中可用。因此,我们需要在开头将其导入。

import falcon
import falcon.asgi

虽然资源类与上一个示例中相同,但必须使用 async 关键字声明 on_get() 方法。我们必须获取 Falcon 的 ASGI 应用程序的实例。

app = falcon.asgi.App()

Example

因此,针对 ASGI 的 hellofalcon.py 如下所示:

import falcon
import falcon.asgi
class HelloResource:
   async def on_get(self, req, resp):
      """Handles GET requests"""
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_TEXT
      resp.text = (
         'Hello World'
      )
app = falcon.asgi.App()
hello = HelloResource()
app.add_route('/hello', hello)

要运行该应用程序,请从命令行启动 Uvicorn 服务器,如下所示:

uvicorn hellofalcon:app –reload

Output

打开浏览器并访问 http://localhost:/8000/hello 。您将在浏览器窗口中看到响应。

asgi

Python Falcon - Uvicorn

Uvicorn 使用 uvloophttptools 库。它还提供对 HTTP/2 和 Websocket 的支持,而 WSGI 无法处理它们。 uvloop 类似于内置 asyncio 事件循环。 httptools 库处理 http 协议。

Falcon 的 ASGI 兼容应用程序在 Uvicorn 服务器上使用以下命令启动:

Uvicorn hellofalcon:app – reload

--reload 选项启用调试模式,以便 app.py 中的任何更改都将自动反映,并且客户端浏览器上的显示将自动刷新。此外,可以使用以下命令行选项:

--host TEXT

将套接字绑定到此主机。[默认 127.0.0.1]

--port INTEGER

将套接字绑定到此端口。[默认 8000]

--uds TEXT

绑定到 UNIX 域套接字。

--fd INTEGER

从该文件描述符绑定到套接字。

--reload

Enable auto-reload.

--reload-dir PATH

明确设置重新加载目录,默认当前工作目录。

--reload-include TEXT

在 watch 时包含文件。默认包含 *.py

--reload-exclude TEXT

在 watch 文件时排除。

--reload-delay FLOAT

前一次和下一次检查之间的延迟(默认 0.25)

--loop [auto

asyncio

uvloop]

事件循环实现。[默认自动]

--http [auto

h11

httptools]

HTTP 协议实现。[默认自动]

--interface auto

asgi

wsgi

选择应用程序接口。[默认自动]

--env-file PATH

Environment configuration file.

--log-config PATH

日志配置文件。支持的格式 .ini, .json, .yaml.

--version

显示 Uvicorn 版本并退出。

--app-dir TEXT

在指定目录(默认当前目录)中查找 APP

--help

显示此消息并退出。

Uvicorn 服务器也可以在程序内部启动,而不是通过上述命令行。要执行该操作,请导入 uvicorn 模块并调用 uvicorn.run() 方法,如下所示 −

import uvicorn
if __name__ == "__main__":
   uvicorn.run("hellofalcon:app", host="0.0.0.0", port=8000, reload=True)

相应地更改 hellofalcon.py 代码,并从命令提示符执行相同操作。可以使用 curl 命令或在浏览器中验证结果,如前所述。

Python Falcon - API Testing Tools

Falcon 是适合开发 API 的极简主义框架。API 是两个应用程序之间的接口。在将其发布到生产环境中之前,API 开发人员需要测试其功能性、可靠性、稳定性、可扩展性和性能等。

为此提供各种 API 测试工具。在本节中,我们将学习如何使用命令行工具 CurlHTTPie ,以及称为 Postman 的 GUI 工具。

cURL

cURL 是一个开源项目,它提供 libcurl 库和一个名为 curl 的命令行工具,可以使用各种协议传输数据。它支持超过 20 种协议,包括 HTTP。cURL 的缩写表示为客户端 URL。从命令行使用 Curl 的语法为 −

curl [options] [URL1, URL2,..]

URL 参数由协议相关的一个或多个 URL 字符串组成。Curl 命令可以使用各种选项来自定义。一些重要的命令行选项如下 −

  1. – X : 提及请求方法。默认情况下,Curl 将 GET 假定为请求方法。要发送 POST、PUT 或 DELTETE 请求,则必须使用此选项。例如 −

Curl –X DELETE http://localhost:8000/student/1
  1. – H : 此选项用于在请求中添加标头。例如 −

Curl –H "Content-Type: application/json" -X GET
http://localhost:8000/students
  1. – i : 当此选项包含在命令行中时,将显示所有响应标头。例如 −

Curl –I –X DELETE http://localhost:8000/student/2
  1. – d : 要在 HTTP 请求中包含要处理的数据,我们必须使用此选项,特别是在需要 POST 或 PUT 请求时。

Curl –H "Content-Type: application/json" -X PUT -d
"{"""marks""":"""50"""}" http://localhost:8000/students/3

HTTPie

HTTPie 是用 Python 编写的命令行工具。据说它是一个“适合人类的 cURL 类似工具”。它支持表单和文件上传,并生成格式良好的彩色终端输出。与 Curl 相比,其富有表现力和直观的语法使它更易于使用。

Examples

  1. GET request − http GET localhost:8000/students

  2. POST request − http POST localhost:8000/students id=4 name="aaa" percent=50

  3. PUT request − http PUT localhost:8000/students/2 id=3 name="Mathews" percent=55

  4. DEETE request − http DELETE localhost:8000/students/2

Postman

Postman 是一款非常流行的 API 测试工具。Gemellini 是一个 GUI 应用程序,它与 Curl 和 HTTPie 相反。它既可用作浏览器插件,也可作为一个桌面应用程序。由于浏览器插件不接受基于本地的 API 请求,我们需要从 https://www.postman.com/downloads. 下载桌面版本。

在完成基于向导的安装后,启动 Postman 应用程序并创建一个新请求。

api1

该下拉框显示可供选择的各种 HTTP 请求类型。

api2

在请求 URL 字段中输入 http://localhost:8000/hello 。右侧的响应窗格显示结果。

api3

我们稍后将在针对 SQLite 数据库测试 Falcon API 的 CRUD 操作时使用对应的请求类型。

Python Falcon - Request & Response

HTTP 协议指出,客户端向服务器发送一个 HTTP 请求,该请求中应用了某些业务逻辑,然后生成响应并将其重定向到客户端。如果在这两个之间进行同步传输,Python 框架会使用 WSGI 标准,而异步传输遵循 ASGI 标准。Falcon 支持这两种标准。

WSGI/ASGI 服务器在上下文数据中提供 Request 和 Response 对象。这些对象被响应者、钩子、中间件等用作参数。对于 WSGI 应用程序,它处理 falcon.Request 类的实例。在 ASGI 应用程序中,它表示 falcon.asgi.Request 类。虽然这两类不同,但它们被设计为具有相似的属性和方法,以便最大限度地减少混淆并实现更简单的移植。

Request

Request 对象表示 HTTP 请求。由于它是由服务器提供的,因此该对象不应由响应者方法直接实例化。此对象提供了以下属性和方法,供响应者、钩子和中间件方法在其中使用 -

  1. method - 请求的 HTTP 方法(例如,“GET”、“POST”等)

  2. host - 主机请求头字段

  3. port - 用于请求的端口。返回给定模式的默认端口(HTTP 为 80,HTTPS 为 443)

  4. uri - 请求的完全限定 URI。

  5. path - 请求 URI 的路径部分(不包括查询字符串)。

  6. query_string - 请求 URI 的查询字符串部分,不包括前面的“?”字符。

  7. cookies - 一个 name/value cookie 对的字典。

  8. content_type - Content-Type 标头或如果标头缺失,则为 None。

  9. stream - 一个用于读取请求正文的文件输入对象(如果存在)。该对象提供对服务器数据流的直接访问,不可寻址。

  10. bounded_stream - 用于流的类文件包装器

  11. headers - 来自请求的原始 HTTP 标头

  12. params - 请求查询参数名称到其值的映射。

  13. get_cookie_values(name) - 为命名 cookie 返回 Cookie 头中提供的所有值。cookies 属性的别名。

  14. get_media() - 以反序列化形式返回请求流。类似于 media 属性。

  15. get_param(name) - 以字符串形式返回查询字符串参数的原始值。如果 application/x-wwwform-urlencoded 媒体类型的 HTML 表单被 POST,Falcon 可以自动从请求正文中解析参数并将它们合并到查询字符串参数中。要启用此功能,请通过 App.req_optionsauto_parse_form_urlencoded 设置为 True。

Response

Response 对象表示服务器对客户端的 HTTP 响应。与 Request 对象一样,Response 对象也不应该由应答者直接实例化。

应答者、hook 函数或中间件方法通过访问以下属性和方法来操作该对象−

  1. status - HTTP 状态代码,例如“200 OK”。这可以被设置成 http.HTTPStatus 的成员、一个 HTTP 状态行字符串或字节字符串,或者一个 int。Falcon 为一些常见状态代码提供了许多常量,这些常量以 HTTP_ 前缀开头,例如 − falcon.HTTP_204

  2. media - 由通过 falcon.RequestOptions 配置的媒体处理程序支持的可序列化对象。

  3. text - 表示响应内容的字符串。

  4. body - text 的已弃用别名。

  5. data - 表示响应内容的字节字符串。

  6. stream - 表示响应内容的文件类似对象。

  7. content_length - 设置 Content-Length 头。在 text 或 data 属性未设置时,它手动设置内容长度。

  8. content_type - 设置 Content-Type 头。Falcon 对常见媒体类型预定义的常量包括 falcon.MEDIA_JSON、falcon.MEDIA_MSGPACK、falcon.MEDIA_YAML、falcon.MEDIA_XML、falcon.MEDIA_HTML、falcon.MEDIA_JS、falcon.MEDIA_TEXT、falcon.MEDIA_JPEG、falcon.MEDIA_PNG 和 falcon.MEDIA_GIF。

  9. append_header (name, value) - 设置或追加此响应的标头。用于设置 cookie。

  10. delete_header (name) - 删除先前为此响应设置的标头。

  11. get_header (name) - 检索给定标头的原始字符串值。

  12. set_cookie (name, value) - 设置响应 cookie。可以多次调用此方法以向响应添加一个或多个 cookie。

  13. set_header (name, value) - 将此响应的标头设置为给定值。

  14. set_stream (stream, content_length) − 设置流和内容长度。

  15. unset_cookie (name, domain=None, path=None) − 清除响应中的 cookie。此方法清除 cookie 的内容,并指示用户代理立即失效其自己的 cookie 副本。

Python Falcon - Resource Class

Falcon 的设计借鉴了 REST 架构样式中的几个关键概念。REST 的含义是 Relational State Transfer 。REST 定义了 Web 应用程序架构的行为方式。

REST 是一种基于资源的架构。在此,REST 服务器承载的所有内容(无论是文件、图像还是数据库中表格中的行)都被视为资源,可能有多种表现形式。REST API 提供对这些资源的受控访问,以便客户端可以检索和修改它们。

服务器上的资源应仅有一个统一资源标识符 (URI)。它只标识资源;它不指定对该资源执行的操作。相反,用户从一组标准方法中进行选择。用于对资源执行操作的 HTTP 动词或方法。POST、GET、PUT 和 DELETE 方法分别执行 CREATE、READ、UPDATE 和 DELETE 操作。

Falcon 使用普通的 Python 类来表示资源。此类在应用程序中充当控制器。它将传入请求转换为一个或多个内部操作,然后根据这些操作的结果将响应返回给客户端。

resource

每个资源类定义各种 "responder" 方法,每个方法对应于资源允许的一种 HTTP 方法。响应程序名称以 "on_" 开头,并且根据它们处理的 HTTP 方法命名,例如 on_get(), on_post(), on_put(), 等。

hellofalcon.py 以上使用的示例代码中, HelloResource (资源类)具有 on_get() 响应程序方法。响应程序必须始终定义至少两个参数来接收请求和响应对象。

import falcon
class HelloResource:
   def on_get(self, req, resp):
      """Handles GET requests"""
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_TEXT
      resp.text = (
         'Hello World'
      )

对于 ASGI 应用程序,响应程序必须是一个协程函数,即必须使用 async 关键字进行定义。

class HelloResource:
   async def on_get(self, req, resp):
      """Handles GET requests"""
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_TEXT
      resp.text = (
         'Hello World'
      )

Request object 表示传入的 HTTP 请求。可以通过此对象访问请求相关的请求头、查询字符串参数和其他元数据。

Response 对象表示应用程序对请求的 HTTP 响应。此对象的属性和方法设置状态、标头和正文数据。它还公开了一个类似 dict 的上下文属性,用于将任意数据传递给挂钩和其他中间件方法。

请注意 HelloResource 在上面的示例中只是一个普通的 Python 类。它可以具有任何名称;但是,惯例是将其命名为 xxxResource

Python Falcon - App Class

此类是基于 Falcon 的 WSGI 应用程序的主要入口点。此类的实例提供了一个可调用的 WSGI 接口和一个路由引擎。

import falcon
app = falcon.App()

此类的 init() 构造函数采用以下关键字参数 −

  1. media_type − 在初始化 RequestOptions 和 ResponseOptions 时要使用的媒体类型。Falcon 允许轻松定制互联网媒体类型处理。默认情况下,Falcon 仅为 JSON 和 HTML(URL 编码和多部分)表单启用处理程序。

  2. request_type − 此参数的默认值是 falcon.Request 类。

  3. response_type − 此参数的默认值是 falcon.Response 类。

为了使 App 对象可调用,其类有一个 call() 方法。

__call__(self, env, start_response)

这是一个 WSGI 应用程序方法。WSGI 开发服务器或其他生产服务器(Waitress/Uvicorn)使用此对象启动服务器实例并侦听来自客户端的请求。

App 类还定义了 add_route() 方法。

add_route(self, uri_template, resource)

此方法有助于将 URI 路径与其资源类中的对象相关联。传入请求按一组 URI 模板路由到资源。如果路径与给定路由的模板匹配,则该请求被传递到相关资源以进行处理。根据请求方法,将调用各自的应答器方法。

Example

让我们向 on_post() 类中添加 HelloResource 应答器方法,并测试 GET 和 POST 请求的端点。

from waitress import serve
import falcon
import json
class HelloResource:
   def on_get(self, req, resp):
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_TEXT
      resp.text = (
         'Hello World'
      )
   def on_post(self, req, resp):
      data=req.media
      nm=data['name']
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_TEXT
      resp.text = (
         'Hello '+nm
      )
app = falcon.App()
hello = HelloResource()
app.add_route('/hello', hello)
if __name__ == '__main__':
   serve(app, host='0.0.0.0', port=8000)

Output

使用 Waitress 服务器运行应用程序,并使用 Curl 检查响应。对于 GET 请求的响应,使用以下命令 −

C:\Users\User>curl localhost:8000/hello
Hello World

我们通过 POST 方法将一些数据发送到 /hello URL,如下所示 −

C:\Users\User>curl -i -H "Content-Type:application/json" -X
POST -d "{"""name""":"""John"""}" http://localhost:8000/hello
HTTP/1.1 200 OK
Content-Length: 10
Content-Type: text/plain; charset=utf-8
Date: Sun, 17 Apr 2022 07:06:20 GMT
Server: waitress
Hello John

要将路由添加到静态文件目录,Falcon 具有 add_static_route() 方法。

add_static_route(self, prefix, directory, downloadable=False,
fallback_filename=None)

prefix 参数是要匹配此路由的路径前缀。directory 参数是要从中提供文件的文件目录。downloadable 参数设置为 True,如果您希望在响应中包含 ContentDisposition 头。 fallback_filename 默认情况下为无,但在请求的文件未找到时可以指定。

add_error_handler() 方法用于为一个或多个异常类型注册处理程序。

add_error_handler(self, exception, handler=None)

可异步调用 App 类具有相同的方法。它在 falcon.asgi 模块中定义。

import falcon.asgi
app=falcon.asgi.App()

请注意,ASGI 应用程序中资源类的响应器必须是 coroutines (使用 async 关键字定义),而不是正常方法。

class HelloResource:
   async def on_get(self, req, resp):
      """Handles GET requests"""
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_TEXT
      resp.text = (
         'Hello World'
      )

Python Falcon - Routing

Falcon 采用 RESTful 架构样式。因此,它使用基于资源的路由。资源类负责通过响应器处理 HTTP 方法,响应器本质上是类方法,名称以 on_ 开头,以小写 HTTP 方法名称结尾(例如, on_get(), on_patch(), on_delete(), 等)。Falcon 应用程序对象的 add_route() 方法将其路由器与其资源类的实例相关联。

在上面使用的 Hellofalcon.py 示例中, on_get()on_post() 响应器在客户端通过 GET 和 POST 方法请求 /hello 路由时被调用。

如果没有路由与请求相匹配,则会引发 HTTPRouteNotFound 的实例。另一方面,如果匹配到路由,但资源并未实现对所请求 HTTP 方法的响应器,则默认响应器将引发 HTTPMethodNotAllowed 的实例。

Field Converters

Falcon 的路由机制允许 URL 向响应器传递参数。URL 包含三个部分:协议(例如 http://https:// ),后跟 IP 地址或主机名。URL 的其余部分在主机名之后的第一个 / 之后称为路径或端点。要传递的参数在端点之后。

routing

它充当资源标识符,例如唯一 ID 或主键。参数名用大括号括起来。路径参数的值除了请求和响应之外,还会进入在响应器方法中定义的参数。

在以下示例中,路由器将资源类对象与其由端点之后的参数组成的 URL 相关联。

from waitress import serve
import falcon
import json
class HelloResource:
   def on_get(self, req, resp, nm):
      """Handles GET requests"""
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_TEXT
      resp.text = (
         'Hello '+nm
      )
app = falcon.App()
hello = HelloResource()
app.add_route('/hello/{nm}', hello)
if __name__ == '__main__':
   serve(app, host='0.0.0.0', port=8000)

我们可以看到, on_get() 响应器方法有一个额外的参数 nm 用于接受从 URL 路由中解析的数据。让我们使用 HTTPie 工具测试 http://localhost:8000/hello/Priya

>http GET localhost:8000/hello/Priya
HTTP/1.1 200 OK
Content-Length: 11
Content-Type: text/plain; charset=utf-8
Date: Mon, 18 Apr 2022 12:27:35 GMT
Server: waitress
Hello Priya

路径参数被解析为的默认数据类型是 str (即字符串)。然而,Falcon 的路由引擎有以下 built-in field converters ,可以使用它们将它们读取到其他数据类型中。

  1. IntConverter − 此类定义在 falcon.routing 模块中。构造函数使用以下参数 −

IntConverter(num_digits=None, min=None, max=None)
app.add_route('/student/{rollno:int(1,1,100}', StudentResource())
  1. UUIDConverter − falcon.routing 模块中的此类将 32 个十六进制数字的字符串转换成 UUID(通用唯一标识符)。

  2. DateTimeConverter − 将参数字符串转换成 datetime 变量。参数必须是 strptime() 函数识别的任何格式的字符串,默认格式是 '%Y-%m-%dT%H:%M:%SZ'

格式字符串使用以下格式代码 −

%a

Abbreviated weekday name

Sun, Mon

%A

Full weekday name

Sunday, Monday

%d

以零填充的十进制数表示的月中天数

01, 02

%-d

以十进制数表示的月中天数

1, 2..

%b

Abbreviated month name

Jan, Feb

%m

以零填充的十进制数表示的月份

01, 02

%B

Full month name

January, February

%-y

以十进制数表示的无世纪年份

0, 99

%Y

以十进制数表示的含世纪年份

2000, 1999

%H

以零填充的十进制数表示的 24 小时时间中的小时数

01, 23

%p

locale’s AM or PM

AM, PM

%-M

以十进制数表示的分钟数

1, 59

%-S

以十进制数表示的秒数

1, 59

在以下示例中, add_route() 函数将一个 URL 与带有两个参数的 Resource 对象关联在一起。默认情况下,第一个参数 nm 为字符串。第二个参数 age 使用 IntConverter

from waitress import serve
import falcon
import json
class HelloResource:
   def on_get(self, req, resp, nm,age):
      """Handles GET requests"""
      retvalue={"name":nm, "age":age}
      resp.body=json.dumps(retvalue)
      resp.status = falcon.HTTP_200
      resp.content_type = falcon.MEDIA_JSON
app = falcon.App()
hello = HelloResource()
app.add_route('/hello/{nm}/{age:int}', hello)
if __name__ == '__main__':
   serve(app, host='0.0.0.0', port=8000)

请注意, on_get() 响应程序使用路径参数形成 dict 对象 - retvalue 。然后,将它的 JSON 表示分配给响应正文的值,并返回给客户端。如前所述,JSON 是 Falcon 响应对象的默认内容类型。

启动 Waitress 服务器,并使用 HTTPie 帮助检查 URL http://localhost:8000/hello/Priya/21 的响应。

http GET localhost:8000/hello/Priya/21
HTTP/1.1 200 OK
Content-Length: 28
Content-Type: application/json
Date: Fri, 22 Apr 2022 14:22:47 GMT
Server: waitress {
   "age": 21,
   "name": "Priya"
}

你还可以使用浏览器中查看响应,如下所示 −

routing hello

Python Falcon - Suffixed Responders

为了理解概念和后缀响应者的需求,让我们定义一个 StudentResource 类。它包含一个 on_get() 响应者,它将学生列表转换 dict 对象为 JSON 并返回作为其响应。

我们还添加 on_post() 响应器,它从传入请求中读取数据并在列表中添加一个新的 dict 对象。

import falcon
import json
from waitress import serve
students = [
   {"id": 1, "name": "Ravi", "percent": 75.50},
   {"id": 2, "name": "Mona", "percent": 80.00},
   {"id": 3, "name": "Mathews", "percent": 65.25},
]
class StudentResource:
   def on_get(self, req, resp):
      resp.text = json.dumps(students)
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON
   def on_post(self, req, resp):
      student = json.load(req.bounded_stream)
      students.append(student)
      resp.text = "Student added successfully."
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_TEXT

使用 Falcon‘s App 对象的 add_route() 函数,我们添加 /students 路由。

app = falcon.App()
app.add_route("/students", StudentResource())

在启动服务器后,我们可以从 HTTPie 命令行测试 GET 和 POST 请求 -

http GET localhost:8000/students
HTTP/1.1 200 OK
Content-Length: 187
Content-Type: application/json
Date: Mon, 18 Apr 2022 06:21:02 GMT
Server: waitress
[
   {
      "id": 1,
      "name": "Ravi",
      "percent": 75.5
   },
   {
      "id": 2,
      "name": "Mona",
      "percent": 80.0
   },
   {
      "id": 3,
      "name": "Mathews",
      "percent": 65.25
   }
]
http POST localhost:8000/students id=4 name="Prachi"
percent=59.90
HTTP/1.1 200 OK
Content-Length: 27
Content-Type: text/plain; charset=utf-8
Date: Mon, 18 Apr 2022 06:20:51 GMT
Server: waitress
Student added successfully.

再次调用 on_get() 来确认添加了新的学生资源。

http GET localhost:8000/students
HTTP/1.1 200 OK
Content-Length: 187
Content-Type: application/json
Date: Mon, 18 Apr 2022 06:21:02 GMT
Server: waitress
[
   {
      "id": 1,
      "name": "Ravi",
      "percent": 75.5
   },
   {
      "id": 2,
      "name": "Mona",
      "percent": 80.0
   },
   {
      "id": 3,
      "name": "Mathews",
      "percent": 65.25
   },
   {
      "id": "4",
      "name": "Prachi",
      "percent": "59.90"
   }
]

在这一阶段,我们希望在 StudentResource 类中有一个 GET 响应器方法,该方法从 URL 中读取 id 参数并从列表中检索相应的 dict 对象。

换句话说,格式为 /student/{id} 的 URL 应当与资源类中的 GET 方法相关联。但显然,一个类不能有两个相同名称的方法。因此,我们定义对 add_route() 方法使用 suffix 参数来区分 on_get() 响应器的两个定义。

通过指定 suffix ='student' ,将一个带有 id 参数的路由添加到 Application 对象。

app.add_route("/students/{id:int}", StudentResource(), suffix='student')

我们现在可以添加 on_get() 方法的另一个定义和此后缀,以便此响应器的名称为 on_get_student() 如下 -

def on_get_student(self, req, resp, id):
   resp.text = json.dumps(students[id-1])
   resp.status = falcon.HTTP_OK
   resp.content_type = falcon.MEDIA_JSON

在添加新路由和 on_get_student() 响应器后启动 Waitress 服务器,并按如下方式测试此 URL -

http GET localhost:8000/students/2
HTTP/1.1 200 OK
Content-Length: 42
Content-Type: application/json
Date: Mon, 18 Apr 2022 06:21:05 GMTy
Server: waitress
{
   "id": 2,
   "name": "Mona",
   "percent": 80.0
}

请注意,当客户端在使用适当的请求头请求 URL 路由 /students/{id:int} 时, on_put() 响应器(用于更新资源)和 on_delete() 响应器(用于删除资源)也将被调用。

我们已经添加了此路由,学生作为后缀。因此, on_put_student() 方法将路径参数解析为整数变量。给定 id 的项目的 JSON 表示形式被获取并使用 PUT 请求中提供的数据进行更新。

def on_put_student(self, req, resp, id):
   student=students[id-1]
   data = json.load(req.bounded_stream)

   student.update(data)
   resp.text = json.dumps(student)
   resp.status = falcon.HTTP_OK
   resp.content_type = falcon.MEDIA_JSON

on_delete_student() 响应器只是删除了 DELETE 请求中指定 id 的项目。返回剩余资源的列表。

def on_delete_student(self, req, resp, id):
   students.pop(id-1)
   resp.text = json.dumps(students)
   resp.status = falcon.HTTP_OK
   resp.content_type = falcon.MEDIA_JSON

我们可以使用 HTTPie 命令测试 API 的 PUT 和 DELETE 操作 -

http PUT localhost:8000/students/2 id=3 name="Mathews"
percent=55
HTTP/1.1 200 OK
Content-Length: 46
Content-Type: application/json
Date: Sat, 18 Apr 2022 10:13:00 GMT
Server: waitress
{
   "id": "3",
   "name": "Mathews",
   "percent": "55"
}
http DELETE localhost:8000/students/2
HTTP/1.1 200 OK
Content-Length: 92
Content-Type: application/json
Date: Sat, 18 Apr 2022 10:18:00 GMT
Server: waitress
[
   {
      "id": 1,
      "name": "Ravi",
      "percent": 75.5
   },
   {
      "id": 3,
      "name": "Mathews",
      "percent": 65.25
   }
]

此 API ( studentapi.py ) 的完整代码如下 -

import falcon
import json
from waitress import serve
students = [
   {"id": 1, "name": "Ravi", "percent": 75.50},
   {"id": 2, "name": "Mona", "percent": 80.00},
   {"id": 3, "name": "Mathews", "percent": 65.25},
]
class StudentResource:
   def on_get(self, req, resp):
      resp.text = json.dumps(students)
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON
   def on_post(self, req, resp):
      student = json.load(req.bounded_stream)
      students.append(student)
      resp.text = "Student added successfully."
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_TEXT
   def on_get_student(self, req, resp, id):
      resp.text = json.dumps(students[id-1])
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON
   def on_put_student(self, req, resp, id):
      student=students[id-1]
      data = json.load(req.bounded_stream)

      student.update(data)

      resp.text = json.dumps(student)
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON
   def on_delete_student(self, req, resp, id):
      students.pop(id-1)
      print (students)
      resp.text = json.dumps(students)
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON
app = falcon.App()
app.add_route("/students", StudentResource())
app.add_route("/students/{id:int}", StudentResource(), suffix='student')
if __name__ == '__main__':
   serve(app, host='0.0.0.0', port=8000)

Python Falcon - Inspect Module

inspect 模块是一个方便的工具,它提供有关注册路由和其他 Falcon 应用程序组件(例如中间件、吸收器等)的信息。

可以通过两种方式检查应用程序 - CLI 工具和以编程方式。从命令行执行 falcon-inspect -tool CLI 脚本,其中给出了申报 Falcon 应用程序对象的 Python 脚本的名称。

例如,如要检查 studentapi.py 中的应用程序对象 -

falcon-inspect-app studentapi:app
Falcon App (WSGI)
Routes:
   ⇒ /students - StudentResource:
   ├── GET - on_get
   └── POST - on_post
   ⇒ /students/{id:int} - StudentResource:
   ├── DELETE - on_delete_student
   ├── GET - on_get_student
   └── PUT - on_put_student

输出显示注册的路由和资源类中的响应程序方法。要以编程方式执行检查,请使用 inspect 模块中的 inspect_app() 函数将应用程序对象作为参数。

from falcon import inspect
from studentapi import app
app_info = inspect.inspect_app(app)
print(app_info)

将上述脚本另存为 inspectapi.py,然后从命令行运行它。

python inspectapi.py
Falcon App (WSGI)
Routes:
   ⇒ /students - StudentResource:
   ├── GET - on_get
   └── POST - on_post
   ⇒ /students/{id:int} - StudentResource:
   ├── DELETE - on_delete_student
   ├── GET - on_get_student
   └── PUT - on_put_student

Python Falcon - Jinja2 Template

Falcon 库主要用于构建 API 和微服务。因此,默认情况下,Falcon 响应器返回 JSON 响应。但是,如果将内容类型更改为 falcon.MEDIA_HTML, ,则可以呈现 HTML 输出。

使用变量数据呈现 HTML 内容非常繁琐。为此,需要使用网络模板库。许多 Python Web 框架都捆绑了特定的模板库。但 Falcon 是一个极简主义微框架,不会捆绑任何模版库。

Jinja2 是许多 Python 框架使用的一个最流行的模板库。在本节中,我们将看到如何在 Falcon 应用中使用 inja2。jinja2 是一种快速且对设计人员友好的模板语言,易于配置和调试。它的沙盒环境便于阻止执行不可信代码、禁止潜在不安全数据,并防止跨站点脚本攻击(称为 XSS attacks )。

jinja2 的另一个非常强大的功能是 template inheritance ,其中您可以定义一个具有通用设计功能的基本模板,而子模板可以覆盖该基本模板。

首先,使用 PIP 实用程序在当前 Python 环境中安装 jinja2

pip3 install jinja2

Hello World Template

jinja2 模块定义了 Template 类。Template 对象是通过读取包含 HTML 脚本(带 .html 扩展名)的文件内容来获取的。通过调用这个 Template 对象的 render() 方法,可以向客户端浏览器呈现 HTML 响应。Response 对象的 content_type 属性必须设置为 falcon.MEDIA_HTML

让我们将以下 HTML 脚本另存为 hello.py 放置在应用文件夹中。

<html>
   <body>
      <h2>Hello World</h2>
   </body>
</html>

Example

下面的资源类中的 on_get() 响应器读取这个文件并将其呈现为 HTML 响应。

import uvicorn
import falcon
import falcon.asgi
from jinja2 import Template
class HelloResource:
   async def on_get(self, req, resp):
      resp.status = falcon.HTTP_200
      resp.content_type = 'text/html'
      fp=open("hello.html","r")
      tempobj=Template(fp.read())
      resp.body=tempobj.render()
app = falcon.asgi.App()
hello = HelloResource()
app.add_route('/hello', hello)
if __name__ == "__main__":
   uvicorn.run("hello:app", host="0.0.0.0", port=8000, reload=True)

Output

运行以上 Python 代码并访问浏览器中的 http://localhost:8000/hello 链接。

jinja2

Template Variable

jinja2 是一个服务器端模板库。网页通过将 jinja2 模板语言的各种元素作为占位符放置到 HTML 脚本内的适当定界符中来构造为模板。模板引擎读取 HTML 脚本,在服务器上用上下文数据替换占位符,重新组装 HTML 并将其呈现给客户端。

Template.render() 函数有一个可选的上下文字典参数。这个字典的关键属性变成模板变量。这有助于呈现响应器在网页中传递的数据。

Example

在以下示例中,路由 /hello/nm 已使用路径参数 nm 注册了资源对象。 on_get() 响应器将其作为上下文传递给从网页获取的模板对象。

import uvicorn
import falcon
import falcon.asgi
from jinja2 import Template
class HelloResource:
   async def on_get(self, req, resp, nm):
      resp.status = falcon.HTTP_200
      resp.content_type = 'text/html'
      fp=open("hello.html","r")
      tempobj=Template(fp.read())
      resp.body=tempobj.render({'name':nm})
app = falcon.asgi.App()
hello = HelloResource()
app.add_route('/hello/{nm}', hello)
if __name__ == "__main__":
   uvicorn.run("hello:app", host="0.0.0.0", port=8000, reload=True)

hello.html 在模板变量名称中读取路径参数。它充当 HTML 脚本中的占位符。它被放置在 {{}} 符号中,以便其值显示为 HTML 响应。

<html>
   <body>
      <h2>Hello {{ name }}</h2>
   </body>
</html>

Output

运行 Python 代码并输入 http://localhost:8000/hello/Priya 作为 URL。浏览器将显示以下输出:

jinja2 hello

Loop in jinja2 Template

如果响应器传递任何 Python 可迭代对象,例如列表、元组或字典,则可以使用其循环构造语法在 jinja2 模板内部遍历其元素。

{% for item in collection %}
HTML block
{% endfor %}

在以下示例中, on_get() 响应器向模板 list.html 发送 students 对象,该对象是 dict 对象的列表。它会遍历数据并将其呈现为 HTML 表格。

import falcon
import json
from waitress import serve
from jinja2 import Template
students = [
   {"id": 1, "name": "Ravi", "percent": 75.50},
   {"id": 2, "name": "Mona", "percent": 80.00},
   {"id": 3, "name": "Mathews", "percent": 65.25},
]
class StudentResource:
   def on_get(self, req, resp):
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_HTML
      fp=open("list.html","r")
      tempobj=Template(fp.read())
      resp.body=tempobj.render({'students':students})

list.html 是一个 jinja2 模板。它接收 students 对象作为字典对象列表,并将每个键的值放在表的 <td>..<.td> 元素内部。

<html>
<body>
<table border=1>
   <thead> <tr>
      <th>Student ID</th> <th>Student Name</th>
      <th>percentage</th>
      <th>Actions</th>
   </tr> </thead>
   <tbody>
   {% for Student in students %}
   <tr> <td>{{ Student.id }}</td> <td>{{ Student.name }}</td>
      <td>{{ Student.percent }}</td>
      <td>
         <a href="#">Edit</a>
         <a href="#">Delete</a>
      </td> </tr>
   {% endfor %}
   </tbody>
</table>
</body>
</html>

访问浏览器地址栏中的 /students 路由。学生的列表将在浏览器中呈现。

jinja2 img

HTML Form Template

在此部分,我们将看到 Falcon 如何从 HTML 表单读取数据。让我们将以下 HTML 脚本保存为 myform.html。我们将使用它来获取 Template 对象并呈现它。

<html>
<body>
   <form method="POST" action="http://localhost:8000/students">
   <p>Student Id: <input type="text" name="id"/> </p>
   <p>student Name: <input type="text" name="name"/> </p>
   <p>Percentage: <input type="text" name="percent"/> </p>
   <p><input type="submit"> </p>
</body>
</html>

Falcon App 对象在 Hello.py 文件中声明,其中还将资源类映射到 /adddnew 路由。 on_get() 响应者读取 myform.html 并对它进行呈现。HTML 表单将会显示出来。通过 POST 方法将表单提交到 /students 路由。

要能够读取表单数据,必须将 auto_parse_form_urlencoded 类的 falcon.RequestOptions 属性设置为 True。

app = falcon.App()
app.req_options.auto_parse_form_urlencoded = True

此处,我们还从 student.py 中导入 StudentResource 类。 on_get() 响应器呈现学生列表。

当用户填写并提交表单时,将调用 on_post() 响应器。此方法收集表单数据到 req.params 属性中,它只不过是一个表单元素及其值的字典。然后附加 students 字典。

def on_post(self, req, resp):
   student=req.params
   students.append(student)

hello.py 的完整代码如下 −

import falcon
import json
from waitress import serve
from jinja2 import Template
from student import StudentResource
class MyResource:
   def on_get(self, req, resp):
      resp.status = falcon.HTTP_200
      resp.content_type = 'text/html'
      fp=open("myform.html","r")
      tempobj=Template(fp.read())
      resp.body=tempobj.render()
app = falcon.App()
app.req_options.auto_parse_form_urlencoded = True
form = MyResource()
app.add_route('/addnew', form)
app.add_route("/students", StudentResource())
if __name__ == '__main__':
   serve(app, host='0.0.0.0', port=8000)

具有 StudentResource 类和 on_get()on_post() 响应器的 student.py 如下 −

import falcon
import json
from waitress import serve
from jinja2 import Template
students = [
   {"id": 1, "name": "Ravi", "percent": 75.50},
   {"id": 2, "name": "Mona", "percent": 80.00},
   {"id": 3, "name": "Mathews", "percent": 65.25},
]
class StudentResource:
   def on_get(self, req, resp):
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_HTML
      fp=open("list.html","r")
      tempobj=Template(fp.read())
      resp.body=tempobj.render({'students':students})

   def on_post(self, req, resp):
      student = req.params
      students.append(student)
      resp.text = "Student added successfully."
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON

从命令行运行 hello.py 。通过输入 http://locLhost:8000/addnew 在浏览器中打开 HTML 表单。

jinja host

将附加 students 数据库字典。访问 /students 路由。你会发现附加了新行。

jinja2 example

Multipart Forms

为了让用户从本地文件系统选择文件,必须将 HTML 表单的 enctype 属性设置为 multipart/form-data。Falcon 使用 MultipartFormHandler 处理 multipart/form-data 媒体类型,允许其迭代表单中的主体部分。

BodyPart 类具有以下属性 −

  1. stream − 仅适用于当前主体部分的流包装器

  2. data − 主体部分内容字节

  3. content_type 如果未指定,将默认为 text/plain,根据 RFC

  4. text − 将当前主体部分解码为文本字符串(仅当它为类型 text/plain 时提供,否则不提供)

  5. media − 由媒体处理程序以与 req.media 相同的方式自动解析

  6. name, filename − 来自 Content-Disposition 标头的相关部分

  7. secure_filename − 已清理的文件名,可以在服务器文件系统安全地使用。

以下 HTML 脚本 ( index.html ) 是一个多部分表单。

<html>
   <body>
      <form action="http://localhost:8000/hello" method="POST" enctype="multipart/form-data">
         <h3>Enter User name</h3>
         <p><input type='text' name='name'/></p>
         <h3>Enter address</h3>
         <p><input type='text' name='addr'/></p>
         <p><input type="file" name="file" /></p>
         <p><input type='submit' value='submit'/></p>
      </form>
   </body>
</html>

此表格由下方代码中的 on_get() 响应器渲染。将表格数据提交给 on_post() 方法,该方法将迭代部分并发送表格数据的 JSON 响应。

import waitress
import falcon
import json
from jinja2 import Template
class HelloResource:
   def on_get(self, req, resp):
      resp.status = falcon.HTTP_200
      resp.content_type = 'text/html'
      fp=open("index.html","r")
      tempobj=Template(fp.read())
      resp.body=tempobj.render()

   def on_post(self, req, resp):
      result=[]
      for part in req.media:
         data={"name" :part.name,
            "content type":part.content_type,
            "value":part.text, "file":part.filename}
         result.append(data)
         resp.text = json.dumps(result)
         resp.status = falcon.HTTP_OK
         resp.content_type = falcon.MEDIA_JSON
app = falcon.App()
hello = HelloResource()
app.add_route('/hello', hello)
if __name__ == '__main__':
   waitress.serve(app, host='0.0.0.0', port=8000)

运行以上程序并访问 http://localhost:8000/hello 链接,如以下所示,以渲染表格 -

jinja2 user

当在填写数据后提交表格时,将在浏览器中显示 JSON 响应,如下所示:

[
   {
      "name": "name",
      "content type": "text/plain",
      "value": "SuyashKumar Khanna",
      "file": null
   },
   {
      "name": "addr",
      "content type": "text/plain",
      "value": "New Delhi",
      "file": null
   },
   {
      "name": "file",
      "content type": "image/png",
      "value": null,
      "file": "hello.png"
   }
]

Python Falcon - Cookies

Cookie 以文本文件形式存储在客户端计算机上。其目的是记住并跟踪与客户端使用相关的数据,以便提供更好的访问者体验和网站统计信息。

请求对象包含 Cookie 的属性。它是客户端已传输的所有 Cookie 变量及其相应值的字典对象。此外,Cookie 还存储其过期时间、路径和网站的域名。

在 Falcon 中,使用 set_cookie() 方法在响应对象上设置 Cookie。

resp.set_cookie('cookiename', 'cookievalue')

此外,Cookie 的参数 max_age (以秒为单位)和域名也可以给出。

import falcon
import json
from waitress import serve
class resource1:
   def on_post(self, req, resp):
      resp.set_cookie("user", 'admin')
      resp.text = "cookie set successfully."
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_TEXT

从命令行调用响应器方法,如下所示:

http POST localhost:8000/cookie
HTTP/1.1 200 OK
Content-Length: 24
Content-Type: text/plain; charset=utf-8
Date: Tue, 26 Apr 2022 06:56:30 GMT
Server: waitress
Set-Cookie: user=admin; HttpOnly; Secure
cookie set successfully.

也可以使用响应对象的 append_header() 方法设置 Cookie Set-cookie 头。

要检索 Cookie,请求对象具有 request.cookies 属性以及 get_cookie_values() 方法。

def on_get(self, req, resp):
   cookies=req.cookies
   values = req.get_cookie_values('user')
   if values:
      v = values[0]
      resp.body={"user":v}
   resp.status = falcon.HTTP_OK
   resp.content_type = falcon.MEDIA_JSON

响应对象的 unset_cookie 方法用于清除当前请求的 Cookie。

resp.unset_cookie('user')

对于 ASGI 应用程序, falcon.asgi.Requestfalcon.Request 实现了相同的 Cookie 方法和属性。 set_cookie()append_header() 的 ASGI 版本是同步的,因此不需要等候。

Python Falcon - Status Codes

默认情况下,HTTP 服务器对客户端请求的响应具有 200 OK 状态。Falcon 提供了自己的状态常量列表,以提高便利性和可读性。

例如,200 OK 状态代码表示为:

resp.status = falcon.HTTP_OK

这些预定义的 Falcon 常量避免了输入错误,并减少了准备响应时必须创建的字符串对象数量。然而,从 Falcon 3.0 版本开始,允许使用 int 代码。

resp.status = 200

对于 ASGI 应用程序,相同的状态代码适用。

Falcon 库中定义的部分状态代码如下:

Informational Codes

  1. HTTP_CONTINUE = HTTP_100

  2. HTTP_SWITCHING_PROTOCOLS = HTTP_101

  3. HTTP_PROCESSING = HTTP_102

Success Status Codes

  1. HTTP_OK = HTTP_200

  2. HTTP_CREATED = HTTP_201

  3. HTTP_ACCEPTED = HTTP_202

  4. HTTP_NON_AUTHORITATIVE_INFORMATION = HTTP_203

  5. HTTP_NO_CONTENT = HTTP_204

  6. HTTP_RESET_CONTENT = HTTP_205

  7. HTTP_PARTIAL_CONTENT = HTTP_206

  8. HTTP_MULTI_STATUS = HTTP_207

  9. HTTP_ALREADY_REPORTED = HTTP_208

  10. HTTP_IM_USED = HTTP_226

Redirection Error Codes

  1. HTTP_MULTIPLE_CHOICES = HTTP_300

  2. HTTP_MOVED_PERMANENTLY = HTTP_301

  3. HTTP_FOUND = HTTP_302

  4. HTTP_SEE_OTHER = HTTP_303

  5. HTTP_NOT_MODIFIED = HTTP_304

  6. HTTP_USE_PROXY = HTTP_305

  7. HTTP_TEMPORARY_REDIRECT = HTTP_307

  8. HTTP_PERMANENT_REDIRECT = HTTP_308

Client Error Codes

  1. HTTP_BAD_REQUEST = HTTP_400

  2. HTTP_UNAUTHORIZED = HTTP_401 # "未验证"

  3. HTTP_PAYMENT_REQUIRED = HTTP_402

  4. HTTP_FORBIDDEN = HTTP_403 # "未授权"

  5. HTTP_NOT_FOUND = HTTP_404

  6. HTTP_METHOD_NOT_ALLOWED = HTTP_405

  7. HTTP_NOT_ACCEPTABLE = HTTP_406

  8. HTTP_PROXY_AUTHENTICATION_REQUIRED = HTTP_407

  9. HTTP_REQUEST_TIMEOUT = HTTP_408

  10. HTTP_CONFLICT = HTTP_409

Server Error Codes

  1. HTTP_INTERNAL_SERVER_ERROR = HTTP_500

  2. HTTP_NOT_IMPLEMENTED = HTTP_501

  3. HTTP_BAD_GATEWAY = HTTP_502

  4. HTTP_SERVICE_UNAVAILABLE = HTTP_503

  5. HTTP_GATEWAY_TIMEOUT = HTTP_504

  6. HTTP_HTTP_VERSION_NOT_SUPPORTED = HTTP_505

  7. HTTP_INSUFFICIENT_STORAGE = HTTP_507

  8. HTTP_LOOP_DETECTED = HTTP_508

  9. HTTP_NETWORK_AUTHENTICATION_REQUIRED = HTTP_511

Python Falcon - Error Handling

为了处理各种错误情况,上述状态代码可用于响应对象。Falcon 还提供了一组错误类。当出现相应的运行时错误情况时,可以引发其对象。

这些错误类源自 HTTPError 类作为其基类。错误对象会引发,如下例所示−

import falcon
class MyResource:
   def on_get(self, req, resp):
      # some Python code
      raise falcon.HTTPBadRequest(
         title="Value Out of Range",
         description="The value is not between permissible range"
      )

Predefined Error Classes

Falcon 提供的一些预定义错误类如下−

  1. HTTPBadRequest − 400 Bad Request。由于客户端错误(例如格式错误的请求语法、无效的请求消息框架等),服务器无法处理请求。

  2. HTTPInvalidHeader − 导致 400 Bad Request,因为请求中的一个标题无效。

  3. HTTPInvalidParam − 表示 400 Bad Request。此错误可能指用请求提交的查询字符串、表单或文档中的无效参数。

  4. HTTPMissingParam − 当请求中缺少参数时,会引发 00 Bad Request。

  5. HTTPForbidden − 服务器理解请求,但拒绝授权该请求。状态代码为 403 Forbidden。

  6. HTTPNotFound − 当服务器没有为目标资源找到当前表示时,则会引发 404 状态代码。它不会表示该表示的缺乏是暂时的还是永久的。

  7. HTTPMethodNotAllowed − 405 Method Not Allowed。目标资源不支持请求行中接收的方法。

  8. HTTPLengthRequired − 当服务器不接受未定义内容长度的请求时。411 Length Required。错误代码。

  9. HTTPUnsupportedMediaType − 如果原始服务器由于有效负荷采用此方法在目标资源上不支持的格式而拒绝为请求提供服务。等效状态代码为 415 Unsupported Media Type。

  10. HTTPUnprocessableEntity − 如果服务器理解请求实体的内容类型且请求实体的语法正确,但无法处理所包含的说明,则引发的错误状态代码为 422 Unprocessable Entity。例如,如果 XML 请求正文包含格式良好的但语义错误的 XML 指令。

  11. HTTPTooManyRequests − 当用户在指定时间内发送了太多请求(“比率限制”)时,会引发 429 Too Many Requests 状态代码。

  12. HTTPInternalServerError − 导致 500 Internal Server Error 的非常常见的错误情况。服务器遇到一种意外情况,阻止它满足请求。

  13. HTTPNotImplemented − 501 (Not Implemented) 状态代码表示服务器不支持满足请求所需的功能。这是在服务器不识别请求方法并且无法为任何资源支持它时适当的响应。

  14. HTTPServiceUnavailable − 503 服务不可用表示服务器由于临时过载或计划维护而当前无法处理请求。

  15. MediaNotFoundError − 400 请求错误。当尝试解析空主体时,媒体处理程序会抛出此异常。

  16. MediaMalformedError − 400 请求错误。当尝试解析格式错误的主体时,媒体处理程序会抛出此异常。

Redirection

也有一组例外,当引发这些异常时,会触发对客户端的重定向响应。状态代码属于 3xx 类型。由以下类表示的这些异常作为 HttpError 的子类来短路请求处理。

  1. HTTPMovedPermanently - 301 永久移动。此状态代码表示目标资源已被分配新的永久 URI。

  2. HTTPFound - 302 发现状态代码,这意味着目标资源临时驻留在其他 URI 下。

  3. HTTPTemporaryRedirect - 此类引发 307(临时重定向)状态代码,这意味着目标资源临时驻留在其他 URI 下,并且用户代理如果执行自动重定向到该 URI,则不能更改请求方法。

  4. HTTPPermanentRedirect - 结果 ib 308 永久重定向,表示目标资源已被分配新的永久 URI。

Python Falcon - Hooks

钩子是在响应客户端请求时在资源类中的特定响应程序方法被调用时自动执行的用户定义函数。Falcon 支持 beforeafter 钩子。

作为钩子使用的函数被定义为具有请求、响应和资源类的参数,除了可能需要的任何可选参数。

def hookfunction(req, resp, resource):
   . . . . .
   . . . . .

此类函数通过应用以下装饰器之一附加到单个响应程序或整个资源类 -

  1. @falcon.before(hookfunction)

  2. @falcon.after(hookfunction)

将 before 钩子应用于 on_post() 响应程序 -

@falcon.before(hookfunction)
def on_post(self, req, resp):
   . . .
   . . .

应用一个 after 钩子 -

@falcon.after(hookfunction)
def on_get(self, req, resp):
   . . .
   . . .

要装饰整个资源类,请在类声明上方使用装饰器 -

@falcon.after(hookfunction)
class SomeResource:
 def on_get(self, req, resp):
   . . .
   . . .
   def on_post(self, req, resp):
   . . .
   . . .

在以下示例中,我们有 StudentResource 类,其中已定义 on_get()on_post() 响应程序。当 POST 请求发送一些数据并且使用它创建的新 dict 对象被添加到 Students 列表中时,会调用 on_post() 响应程序。

接收到的数据需要在处理之前进行验证。为此,已定义以下函数。它检查 percent 参数的值是否在 0 和 100 之间。仅当数据通过此条件时,才将其传递给响应程序。

def checkinput(req, resp, resource,params):
   student = json.load(req.bounded_stream)
   if "name" not in student:
      raise falcon.HTTPBadRequest(
         title="Bad request", description="Bad input, name must be provided."
      )

   per=int(student['percent'])
   if per<0 or per>100:
      raise falcon.HTTPBadRequest(
         title="Bad request", description="Bad input, invalid percentage"
      )
      req.context.data = student

此函数作为 StudentResource 类的 on_post() 响应程序上的钩子应用。

import falcon
import json
from waitress import serve
students = [
   {"id": 1, "name": "Ravi", "percent": 75.50},
   {"id": 2, "name": "Mona", "percent": 80.00},
   {"id": 3, "name": "Mathews", "percent": 65.25},
]
class StudentResource:
   def on_get(self, req, resp):
      resp.text = json.dumps(students)
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON
   @falcon.before(checkinput)
   def on_post(self, req, resp):
      student = json.load(req.context.data)
      students.append(student)
      resp.text = "Student added successfully."
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_TEXT

   def on_get_student(self, req, resp, id):
      resp.text = json.dumps(students[id-1])
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON
app = falcon.App()
app.add_route("/students", StudentResource())
if __name__ == '__main__':
   serve(app, host='0.0.0.0', port=8000)

让我们运行 Waitress 服务器并发起 POST 请求。

http POST localhost:8000/students id=4 percent=50
HTTP/1.1 400 Bad Request
Content-Length: 76
Content-Type: application/json
Date: Tue, 26 Apr 2022 14:49:07 GMT
Server: waitress
Vary: Accept {
   "description": "Bad input, name must be provided.",
   "title": "Bad request"
}

由于数据不包含 name 参数的值,因此引发异常。

在下面显示的另一个 POST 请求中,percent 参数的值未满足所需条件,因此会引发异常。

http POST localhost:8000/students id=4 name="aaa" percent=500
HTTP/1.1 400 Bad Request
Content-Length: 72
Content-Type: application/json
Date: Tue, 26 Apr 2022 15:01:20 GMT
Server: waitress
Vary: Accept {
   "description": "Bad input, invalid percentage",
   "title": "Bad request"
}

Python Falcon - Middleware

"middleware" 是一个函数,用它来处理每个请求(在由任何特定响应程序处理之前),以及在返回每个响应之前。此函数会处理发送到应用程序的每个请求。

中间件的作用类似于 hook。但是,与 hook 不同,中间件方法全局应用于整个应用程序。它可以通过运行其中定义的代码对请求执行某些流程,然后将请求传递给相应的操作函数进行处理。它还可以在返回生成的响应之前对其进行处理。

中间件是一个类,它实现一个或多个以下事件处理程序方法。对于 WSGI 应用程序,方法为:

  1. process_request (self, req, resp) - 此方法在将请求路由之前对其进行处理。

  2. process_resource (self, req, resp, resource, params) - 在路由后处理请求。可以传递一个 dict 对象,该对象表示从路由的 URI 模板字段派生的任何其他参数。

  3. process_response (self, req, resp, resource, req_succeeded) - 此方法用于响应后处理(路由后)。如果未引发异常,则 req_succeeded 参数为 True,否则为 False。

对于 ASGI 应用程序,除了上述方法外,中间件类还可以定义更多的方法。

为了考虑 ASGI 规范的可选部分寿命事件,可以包括启动和关闭事件处理程序。

  1. process_startup (self, scope, event) - 此方法处理 ASGI 寿命启动事件。当服务器准备好启动并接受连接时调用它,但在开始这样做之前。

  2. process_shutdown(self, scope, event) - 此方法处理 ASGI 寿命关闭事件。当服务器停止接受连接并关闭所有活动连接时调用它。

由于 ASGI 应用程序还响应 Websocket 协议下的请求,因此中间件可以定义以下协程方法:

  1. process_request_ws (self, req, ws) - 此方法在将 WebSocket 握手请求路由之前对其进行处理。

  2. process_resource_ws (self, req, ws, resource, params) - 此方法在路由后处理 WebSocket 握手请求。可以将从路由的 URI 模板字段派生的 dict 对象传递给资源的响应程序。

中间件类的实例必须在初始化时添加到 Falcon 应用程序对象中。对于 WSGI Falcon 应用程序:

class MyMiddleware:
   def process_request(self, req, resp):
      pass
   def process_resource(self, req, resp, resource, params):
      pass
   def process_response(self, req, resp, resource, req_succeeded):
      pass
from falcon import App
app=App(middleware=[MyMiddleware()])

对于 ASGI 应用程序:

class MyMiddleware:
   async def process_startup(self, scope, event):
      pass
   async def process_shutdown(self, scope, event):
      pass
   async def process_request(self, req, resp):
      pass
   async def process_resource(self, req, resp, resource, params):
      pass
   async def process_response(self, req, resp, resource, req_succeeded):
      pass
   async def process_request_ws(self, req, ws):
      pass
   async def process_resource_ws(self, req, ws, resource, params):
      pass
from falcon.asgi import App
app=App(middleware=[MyMiddleware()])

Python Falcon - CORS

"Cross-Origin Resource Sharing" (CORS) 是指运行在某个客户端浏览器上的前端应用程序尝试通过 JavaScript 代码与后端通信的情况,并且后端与前端处于不同的“来源”中。这里的来源是协议、域名和端口号的组合。因此, http://localhosthttps://localhost 有不同的来源。

如果具有一个来源的 URL 的浏览器发送从另一个来源执行 JavaScript 代码的请求,则浏览器会发送 OPTIONS http 请求。如果后端通过发送适当的标头授权来自此其他来源的通信,它将允许前端的 JavaScript 将其请求发送到后端。

若要为所有响应启用 CORS 策略,则应将 Falcon 应用配置如下所示:

from falcon import App
app=App(cors_enable=True)

若要明确指定允许的源,请导入 CORSMiddleware ,并将源列表添加到应用中间件中,同时附带各自的凭据。

from falcon import App
app = falcon.App(middleware=falcon.CORSMiddleware(allow_origins='example.com', allow_credentials='*')

Python Falcon - Websocket

WebSocket 是客户端与服务器之间的持久连接,可在这两者之间提供双向 full-duplex 通信。通信经由 HTTP 在单个 TCP/IP 套接字连接中发生。这可以看做是 HTTP 的升级,而不是一个协议本身。

HTTP 的局限性之一是它是一个严格的半双工或单向协议。另一方面,借助 WebSocket,我们可以发送基于消息的数据(类似于 UDP),但借助 TCP 的可靠性。WebSocket 使用 HTTP 作为初始传输机制,但在收到 HTTP 响应后保持 TCP 连接的活动状态。可以将同一连接对象用于客户端和服务器之间的双向通信。因此,可以使用 WebSocket API 构建实时应用。

Falcon 的 Websocket 支持仅适用于 ASGI 应用。若要提供 Websocket 能力,资源类应具有 on_websocket() 回应器协程。

async def on_websocket(self, req, ws):
   . . .

Websocket 请求还可由钩子和中间件拦截。将传递 falcon.asgi.WebSocket 对象,而不是 Response 对象。

How Does a WebSocket Function in Falcon?

以下示例演示了 Falcon 应用中 WebSocket 的功能。首先,我们有一个 on_get() 回应器,它用来呈现一个模板。

Example

客户端浏览器显示一个带有文本字段和按钮的表单,单击该按钮时,就会创建一个 websocket 对象,并触发 on_websocket() 回应器。它接受用户输入的消息,并以“消息文本为”为前缀将它反馈给客户端。

import falcon
import falcon.asgi
import jinja2
html = """
<!DOCTYPE html>
<html>
   <head>
      <title>Chat</title>
   </head>
   <body>
      <script>
         var ws = new WebSocket("ws://localhost:8000/hello");
         ws.onmessage = function(event) {
            var messages =document.getElementById('messages')
            var message = document.createElement('li')
            var content = document.createTextNode(event.data)
            message.appendChild(content)
            messages.appendChild(message)
         };
         function sendMessage(event) {
            var input = document.getElementById("messageText")
            ws.send(input.value)
            input.value = ''
            event.preventDefault()
         }
      </script>
      <h1>WebSocket Chat</h1>
      <form action="" onsubmit="sendMessage(event)">
         <input type="text" id="messageText" autocomplete="off"/>
         <button>Send</button>
      </form>
      <ul id='messages'></ul>
   </body>
</html>
"""
class HelloResource:
   async def on_get(self, req, resp):
      """Handles GET requests"""
      resp.status = falcon.HTTP_200
      resp.content_type = 'text/html'
      template=jinja2.Template(html)
      resp.body=template.render()
   async def on_websocket(self, req, websocket):
      await websocket.accept()
      while True:
         data = await websocket.receive_text()
         await websocket.send_text(f"Message text was: {data}")
app = falcon.asgi.App()
hello = HelloResource()
app.add_route('/hello', hello)
import uvicorn
if __name__ == "__main__":
   uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Output

启动 Uvicorn 服务器,并访问 http://localhost:8000/ws URL 以显示聊天表单。

websocket img

输入一些文本并按 Send 按钮。

websocket example

Python Falcon - SQLAlchemy Models

若要演示 Falcon 的响应器如何发挥作用 ( on_post(), on_get(), on_put()on_delete() ),我们对 Python 词典对象的列表形式的内存数据库执行了 CRUD (表示创建、检索、更新和删除)操作。相反,我们可以使用任何关系数据库(例如 MySQL、Oracle 等)来执行存储、检索、更新和删除操作。

我们不会使用 DB-API 合规的数据库驱动程序,而会使用 SQLAlchemy 作为 Python 代码和数据库之间的界面(我们将使用 SQLite 数据库,因为 Python 已内置对它的支持)。SQLAlchemy 是一个流行的 SQL 工具包和 Object Relational Mapper

对象关系映射是一种编程技术,用于在面向对象编程语言的不同类型系统之间转换数据。通常,像 Python 这种面向对象语言中使用的类型系统包含非标量类型。但是,大多数数据库产品的(例如 Oracle、MySQL 等)数据类型都是基本类型,例如整数和字符串。

在 ORM 系统中,每个类都映射到底层数据库中的一个表。ORM 负责处理这些问题,让你专注于对系统逻辑进行编程,而不再需要自己编写乏味的数据库接口代码。

为了使用 SQLALchemy,我们需要先使用 PIP 安装程序来安装该库。

pip install sqlalchemy

SQLAlchemy 被设计为使用专为某个特定数据库而构建的 DBAPI 实现来运行。它使用方言系统与各种类型的 DBAPI 实现和数据库通信。所有方言都要求已安装相应的 DBAPI 驱动程序。

以下是包含的方言:

  1. Firebird

  2. Microsoft SQL Server

  3. MySQL

  4. Oracle

  5. PostgreSQL

  6. SQLite

  7. Sybase

Database Engine

由于我们打算使用 SQLite 数据库,因此需要为名为 test.db 的数据库创建一个数据库引擎。从 sqlalchemy 模块中导入 create_engine() 函数。

from sqlalchemy import create_engine
from sqlalchemy.dialects.sqlite import *
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args =
{"check_same_thread": False})

为了与数据库交互,我们需要获取它的句柄。会话对象是数据库的句柄。会话类使用 sessionmaker() 定义,一个可配置的会话工厂方法,它绑定到引擎对象。

from sqlalchemy.orm import sessionmaker, Session
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

接下来,我们需要一个声明式基类,用于在声明式系统中存储类的目录和映射的表。

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

Model class

Students 是 Base 的子类,映射到了数据库中的 students 表。Books 类中的属性对应于目标表中列的数据类型。请注意,id 属性对应于 book 表中的主键。

class Students(Base):
   __tablename__ = 'student'
   id = Column(Integer, primary_key=True, nullable=False)
   name = Column(String(63), unique=True)
   marks = Column(Integer)
Base.metadata.create_all(bind=engine)

create_all() 方法创建了数据库中对应的表。可以通过使用像 SQLiteStudio 这样的 SQLite 视觉工具来确认。

sqlite

我们现在需要声明一个 StudentResource 类,其中定义了 HTTP 响应者方法,用于对学生表执行 CRUD 操作。此类的对象与路由关联,如下图段所示:

import falcon
import json
from waitress import serve
class StudentResource:
   def on_get(self, req, resp):
      pass
   def on_post(self, req, resp):
      pass
   def on_put_student(self, req, resp, id):
      pass
   def on_delete_student(self, req, resp, id):
      pass
app = falcon.App()
app.add_route("/students", StudentResource())
app.add_route("/students/{id:int}", StudentResource(), suffix='student')

on_post()

其余代码与内存中 CRUD 操作类似,不同之处在于操作函数通过 SQLalchemy 接口与数据库进行交互。

on_post() 响应者方法首先根据请求参数构造 Students 类的对象,并将其添加到 Students 模型中。由于此模型映射到了数据库中的 students 表,因此会添加相应的行。 on_post() 方法如下:

def on_post(self, req, resp):
   data = json.load(req.bounded_stream)
   student=Students(id=data['id'], name=data['name'], marks=data['marks'])
   session.add(student)
   session.commit()
   resp.text = "Student added successfully."
   resp.status = falcon.HTTP_OK
   resp.content_type = falcon.MEDIA_TEXT

如前所述,当收到 POST 请求时,会调用 on_post() 响应者。我们将使用 Postman 应用来传递 POST 请求。

启动 Postman,选择 POST 方法并传递值 (id=1, name="Manan" and marks=760 作为 body 参数。请求成功处理,并向 students 表中添加一行。

postman

继续并发送多个 POST 请求以添加记录。

on_get()

此响应者用于检索 Students 模型中的所有对象。 Session 对象上的 query() 方法检索对象。

rows = session.query(Students).all()

由于 Falcon 响应者的默认响应采用 JSON 格式,因此我们必须将上述查询的结果转换成 dict 对象的列表。

data=[]
for row in rows:
   data.append({"id":row.id, "name":row.name, "marks":row.marks})

StudentResource 类中,我们添加执行此操作并发送其 JSON 响应的方法 on_get() ,如下所示:

def on_get(self, req, resp):
   rows = session.query(Students).all()
   data=[]
   for row in rows:
      data.append({"id":row.id, "name":row.name, "marks":row.marks})
      resp.text = json.dumps(data)
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_JSON

可以在 Postman 应用中测试 GET 请求操作。 /students URL 将显示 JSON 响应,显示学生模型中所有对象的数据。

postman example

Postman 应用的结果窗格中显示的两条记录也可以在 SQLiteStudio 的数据视图中进行验证。

sqllite1

on_put()

on_put() 响应者执行 UPDATE 操作。它响应 URL /students/id 。要从 Students 模型中获取具有给定 id 的对象,我们对查询结果应用过滤器,并使用从客户端接收到的数据更新其属性的值。

student = session.query(Students).filter(Students.id == id).first()

on_put() 方法的代码如下:

def on_put_student(self, req, resp, id):
   student = session.query(Students).filter(Students.id == id).first()
   data = json.load(req.bounded_stream)
   student.name=data['name']
   student.marks=data['marks']
   session.commit()
   resp.text = "Student updated successfully."
   resp.status = falcon.HTTP_OK
   resp.content_type = falcon.MEDIA_TEXT

让我们在 Postman 的帮助下更新 Students 模型中 id 为 id=2 的对象,并更改 name 和 marks。请注意,这些值作为 body 参数传递。

onget

SQLiteStudio 中的数据视图显示这些修改已经生效。

onput

on_delete()

最后,DELETE 操作很简单。我们需要获取给定 id 的对象,并调用 delete() 方法。

def on_delete_student(self, req, resp, id):
   try:
      session.query(Students).filter(Students.id == id).delete()
      session.commit()
   except Exception as e:
      raise Exception(e)
      resp.text = "deleted successfully"
      resp.status = falcon.HTTP_OK
      resp.content_type = falcon.MEDIA_TEXT

为了测试 on_delete() 应答者,让我们在 Postman 的帮助下删除 id=2 的对象,如下所示 −

ondelete

Python Falcon - Testing

Falcon 的测试模块是 Falcon 应用程序的功能测试框架。它包含各种测试类和实用程序函数,以支持功能测试。测试框架同时支持 unittestpytest

我们将使用以下脚本 ( myapp.py ) 来演示测试功能。它包含一个 HelloResource 类,其中包含一个 on_get() 应答者,该应答者呈现一个 Hello World 的 JSON 响应。 create() 函数返回 Falcon 的 Application 对象,其中添加了一个使用 '/' URL 注册的路由。

from waitress import serve
import falcon
import json
class HelloResource:
   def on_get(self, req, resp):
      """Handles GET requests"""
      resp.text=json.dumps({"message":"Hello World"})

   # This is the default status
   resp.status = falcon.HTTP_200

   # Default is JSON, so override
   resp.content_type = falcon.MEDIA_JSON
def create():
   app = falcon.App()
   hello = HelloResource()
   app.add_route('/', hello)
   return app
app=create()
if __name__ == '__main__':
   serve(app, host='0.0.0.0', port=8000)

Using unittest

testing.TestCase 扩展了 unittest ,以方便对使用 Falcon 编写的 WSGI/ASGI 应用程序进行功能测试。我们需要从这个基类继承并编写测试。

TestCase 子类中的测试函数的名称为 simulate_ (),其中 *' '* 表示 HTTP 方法,例如 GET、POST 等。这意味着我们必须获取 simulate_get() 函数的结果,并通过断言函数将其与预期结果进行比较。

simulate_ ()* 函数接收两个参数。

simulate_*(app, route)

以下是 test-myapp.py 的代码。它执行 simulate_get() 函数并将其结果与预期结果进行断言,并指示测试是否失败或通过。

from falcon import testing
import myapp
class MyTestCase(testing.TestCase):
   def setUp(self):
      super(MyTestCase, self).setUp()
      self.app = myapp.create()
class TestMyApp(MyTestCase):
   def test_get_message(self):
      doc = {'message': 'Hello world!'}
      result = self.simulate_get('/')
      self.assertEqual(result.json, doc)
if '__name__'=='__main__':
   unittest.main()

使用以下命令运行上述测试 −

python -m unittest test-myapp.py
F
==============================================================
FAIL: test_get_message (test-myapp.TestMyApp)
--------------------------------------------------------------
Traceback (most recent call last):
   File "E:\falconenv\test-myapp.py", line 17, in test_get_message
   self.assertEqual(result.json, doc)
AssertionError: {'message': 'Hello World'} != {'message':
'Hello world!'}
- {'message': 'Hello World'}
? ^
+ {'message': 'Hello world!'}
? ^ +
--------------------------------------------------------------
Ran 1 test in 0.019s
FAILED (failures=1)

Using Pytest

要使用 PyTest 框架进行测试,你需要使用 PIP 实用程序安装它。

pip3 install pytest

要运行 test 函数,我们需要一个 testing.TestClient 类的对象。它模拟 WSGI 和 ASGI 应用程序的请求。这个对象首先通过将 Falcon 应用程序对象作为参数来获取。

我们运行 simulate_ ()* 函数并断言其结果是否与预期输出一致,以确定测试是否失败或通过。在这两个示例中,测试都失败了,因为 Hello World 消息中的“W”不同。应答者返回大写“W”,而测试函数中有小写“W”。

from falcon import testing
import pytest
import myapp
@pytest.fixture()
def client():
   return testing.TestClient(myapp.create())
def test_get_message(client):
   doc = {'message': 'Hello world!'}
   result = client.simulate_get('/')
   assert result.json == doc

使用以下命令运行上述测试 −

pytest test-myapp.py –v
=========== test session starts ==========================
platform win32 -- Python 3.8.6, pytest-7.1.2, pluggy-1.0.0 --
e:\falconenv\scripts\python.exe
cachedir: .pytest_cache
rootdir: E:\falconenv
plugins: anyio-3.5.0
collected 1 item
test-myapp.py::test_get_message FAILED
[100%]
==================== FAILURES =======================
_____________________________________________________
test_get_message
_____________________________________________________
client = <falcon.testing.client.TestClient object at 0x0000000003EAA6A0>
def test_get_message(client):
   doc = {'message': 'Hello world!'}
   result = client.simulate_get('/')
> assert result.json == doc
E AssertionError: assert {'message': 'Hello World'} ==
{'message': 'Hello world!'}
E Differing items:
E {'message': 'Hello World'} != {'message': 'Hello world!'}
E Full diff:
E - {'message': 'Hello world!'}
E ? ^ -
E + {'message': 'Hello World'}
E ? ^
test-myapp.py:42: AssertionError
============ short test summary info ==================
FAILED test-myapp.py::test_get_message - AssertionError:
assert {'message': 'Hello World'} == {'message': 'Hello
world!'}
============ 1 failed in 4.11s ========================

Python Falcon - Deployment

可以使用启用 mod_wsgi 模块的 Apache 服务器部署 Falcon Web 应用程序,就像任何 WSGI 应用程序一样。另一种方法是使用 uWSGIgunicorn 进行部署。

uWSGI 是一个快速且高度可配置的 WSGI 服务器。如果与 NGNIX 一起使用,它将以速度的形式在生产就绪环境中提供更好的性能。

首先,在 Python 虚拟环境中使用 PIP 安装程序安装 Falcon 和 uWSGI,并使用 wsgi.py 将 Falcon 的应用程序对象公开给 uWSGI,如下所示 −

import os
import myapp
config = myproject.get_config(os.environ['MYAPP_CONFIG'])
application = myapp.create(config)

要配置 uWSGI,请准备一个 uwsgi.ini 脚本,如下所示 −

[uwsgi]
master = 1
vacuum = true
socket = 127.0.0.1:8080
enable-threads = true
thunder-lock = true
threads = 2
processes = 2
virtualenv = /path/to/venv
wsgi-file = venv/src/wsgi.py
chdir = venv/src
uid = myapp-runner
gid = myapp-runner

你现在可以像这样启动 uWSGI −

venv/bin/uwsgi -c uwsgi.ini

虽然 uWSGI 可以直接处理 HTTP 请求,但使用 NGINX 等反向代理可能会很有帮助。NGINX 本地支持 uWSGI 协议,以便有效地将请求代理到 uWSGI。

安装 Ngnix,然后创建一个 NGINX conf 文件,如下所示 −

server {
   listen 80;
   server_name myproject.com;
   access_log /var/log/nginx/myproject-access.log;
   error_log /var/log/nginx/myproject-error.log warn;
   location / {
      uwsgi_pass 127.0.0.1:8080
      include uwsgi_params;
   }
}

终于启动了 Ngnix 服务器。您应该有正在运行的工作应用程序。