Python Falcon 快速指南



Python Falcon - 簡介

Falcon 是一個用於開發關鍵型 REST API 和微服務的 Python 庫。它支援 WSGI 和 ASGI 規範。Falcon 框架由 Kurt Griffiths 於 2013 年 1 月開發。Falcon 的最新版本是 3.1.0,於 2022 年 3 月釋出。

Falcon 是一個輕量級的 Web 開發框架。其極簡主義的設計允許開發者根據需要選擇最佳策略和第三方包。

Falcon - 重要特性

Falcon 在 Apache 2.0 許可證條款下發布。

Falcon 的一些重要特性包括:

  • 最新版本的 Falcon 支援 ASGI、WSGI 和 WebSocket。

  • Falcon 提供對 asyncio 的原生支援。

  • 其穩定的介面確保向後相容性。

  • Falcon 遵循 REST 架構風格來構建 API。

  • 基於類的 HTTP 資源構建。

  • 高度最佳化的、可擴充套件的程式碼庫。

  • Falcon 透過請求和響應類提供對標頭和正文的輕鬆訪問。

  • 提供中介軟體元件和鉤子用於 DRY 請求處理。

  • 慣用的 HTTP 錯誤響應和異常處理。

Falcon - 設計理念

Falcon 最大限度地減少了物件的例項化數量,以避免建立物件的開銷並減少記憶體使用。同一例項將用於服務到達該路由的所有請求。

  • 異常由資源響應器(例如 on_get()、on_post() 等方法)正確處理。Falcon 不會努力保護響應器程式碼免受自身影響。高質量的 Falcon API 應滿足以下要求:

    • 資源響應器將響應變數設定為合理的值。

    • 您的程式碼經過充分測試,具有較高的程式碼覆蓋率。

    • 每個響應器中都提供了自定義錯誤處理程式來預測、檢測和處理錯誤。

  • Falcon 框架是執行緒安全的。為每個傳入的 HTTP 請求建立單獨的新請求和響應物件。但是,附加到路由的每個資源類的單個例項在所有請求之間共享。中介軟體物件、鉤子和自定義錯誤處理程式也是共享的。因此,您的 WSGI 應用整體將是執行緒安全的。

  • 從 3.0 版本開始,Falcon 支援 asyncio。使用 falcon.asgi.App 類建立非同步應用程式,並透過 ASGI 應用程式伺服器(例如 Uvicorn)提供服務。

  • Falcon 的非同步版本支援 ASGI WebSocket 協議。

Falcon - 與其他框架的比較

Python Web 框架主要分為兩大類:全棧式微型框架。

  • 全棧式框架帶有內建功能和庫。Django、TurbogearsWeb2Py 是全棧式框架。

  • 相比之下,微型框架非常簡潔,只提供最基本的功能;因此,它使開發者可以自由選擇官方或第三方的擴充套件,並且只包含他們需要的外掛。Flask、Falcon、Pyramid 屬於微型框架類別。

我們根據以下引數比較 Falcon 框架與不同的框架:

效能

與 Flask 和 Pyramid 等微型框架相比,Falcon 應用程式非常快。全棧式框架通常較慢。

REST 支援

Falcon 旨在成為開發 REST API 和微服務的首選框架。FastAPI 也鼓勵 REST 開發。Flask 和 Django 沒有內建的 REST 支援。但是,可以使用擴充套件來啟用它。

模板引擎

Falcon 應用不應提供模板網頁。它沒有捆綁任何模板庫。但是,可以使用 jinja2Mako 庫。另一方面,Flask 內建支援 jinja2。Django 有自己的模板庫。FastAPI 也能處理任何選擇的模板庫。

資料庫支援

在 Falcon 中,資料庫支援不是內建的。可以使用 SQLAlchemy 模型與關係資料庫(如 MySQL、PostgreSQL、SQLite 等)進行互動。另一方面,Django 有自己的 ORM 框架,可以直接使用。

Flask 應用程式也可以透過 Flask 擴充套件與資料庫互動。早期版本的 TurboGears 與 SQLObject ORM 庫相容。較新的版本與 SQLAlchemy 相容。

靈活性

Falcon 應用程式非常靈活。它非常適合需要高度自定義和效能調整的應用程式。FastAPI 和 Flask 也易於編寫程式碼,並且不會限制使用者使用特定的專案或程式碼佈局。

安全性

Falcon 沒有內建的安全支援。Django 和 FastAPI 等其他框架確保高度安全性。Flask 也提供了針對 CSRF 和 XSS 等安全威脅的出色保護。

測試

Falcon 使用 unittest 和 Pytest 提供內建測試支援。Flask 和 Django 也支援 unittest。FastAPI 支援 unittest 和 starlette 測試功能。

Python Falcon - 環境搭建

最新版本的 Falcon 需要 Python 3.5 或更高版本。安裝 Falcon 最簡單也是推薦的方法是使用 PIP 安裝程式,最好是在虛擬環境中。

可以透過執行以下命令安裝最新穩定版本:

pip3 install falcon

要驗證安裝是否成功,請匯入庫並檢查其版本。

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

要安裝最新的 Beta 版本,應使用以下命令:

pip3 install --pre falcon

從早期版本開始,Falcon 就支援 WSGI。可以使用 Python 標準庫模組 wsgiref 中的內建 WSGI 伺服器執行 Falcon 應用程式。但是,它不適合生產環境,為此需要 gunicorn、waitress 或 uwsgi 等 WSGI 伺服器。

對於 Windows 上的 Falcon,可以使用 Waitress,這是一個生產級的純 Python WSGI 伺服器。像往常一樣,使用 pip 安裝程式安裝它。

pip3 install waitress

Gunicorn 伺服器無法安裝在 Windows 上。但是,它可以在 Windows 10 上的 Windows Subsystem Linux (WSL) 環境中使用。要在 Linux、WSL 或 Docker 容器內使用 gunicorn,請使用

pip3 install gunicorn

如果要執行非同步 Falcon 應用程式,則需要 ASGI 相容的應用程式伺服器。Uvicorn 伺服器可以在 Windows 和 Linux 系統上使用。

pip3 install uvicorn

Python Falcon - WSGI vs ASGI

Web 伺服器閘道器介面 (WSGI)

一些最流行的 Python Web 框架實現了 WSGI(代表 Web Server Gateway Interface)。WSGI 本質上是一組規範,用於 Web 伺服器和 Web 應用程式之間的通用介面,由 Web 伺服器軟體實現,用於處理來自基於 Python 的 Web 應用程式的請求。WSGI 規範於 2003 年(PEP 333)首次推出,後來於 2010 年(PEP 3333)更新。

伺服器透過傳遞以下引數來呼叫 WSGI 應用程式物件:

  • environ - 一個類似於 CGI 環境變數和某些 WSGI 特定變數的 Python dict 物件。

  • 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 函式成為 協程,因此不能像普通函式一樣執行。相反,我們需要使用 asyncio.run(coroutine) 來呼叫它。協程的執行可以透過 await 關鍵字暫停,直到另一個協程完成。

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

asyncio.run(main())

非同步伺服器閘道器介面 (ASGI)

ASGI 代表 Asynchronous Server Gateway Interface(根據其官方文件,它是 WSGI 的繼承者),它為 Python Web 伺服器、應用程式和框架添加了非同步功能。

ASGI 應用程式是一個非同步可呼叫物件(使用者定義的函式或具有 __call__() 方法的類的物件)。它接受三個引數,如下所示:

  • Scope - 包含特定連線詳細資訊的 dict

  • Send - 一個非同步可呼叫物件,應用程式可以使用它向客戶端傳送事件訊息。

  • 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()

示例

讓我們將所有這些程式碼片段放在 **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...

輸出

在另一個終端中,執行 Curl 命令如下:

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

您也可以開啟瀏覽器視窗並輸入上述 URL 以獲得“Hello World”響應。

Python Falcon 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'

示例

在 **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** 並訪問瀏覽器中的 **https://:8000/hellolink**,如前所述。請注意,host 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 應用程式伺服器。流行的選擇包括:

  • Uvicorn
  • Daphne
  • Hypercorn

在本教程中,我們將使用 **Uvicorn** 伺服器進行 **async** 示例。

Hello World - ASGI

Falcon 的 ASGI 相關功能在 falcon.asgi 模組中可用。因此,我們需要在開始時匯入它。

import falcon
import falcon.asgi

雖然資源類與之前的示例相同,但 on_get() 方法必須使用 async 關鍵字宣告。我們必須獲取 Falcon 的 ASGI 應用程式的例項。

app = falcon.asgi.App()

示例

因此,用於 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

輸出

開啟瀏覽器並訪問 **https://:8000/hello**。您將在瀏覽器視窗中看到響應。

ASGI

Python Falcon - Uvicorn

Uvicorn 使用 **uvloop** 和 **httptools** 庫。它還提供對 HTTP/2 和 WebSockets 的支援,而 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 啟用自動重新載入。
--reload-dir PATH 顯式設定重新載入目錄,預設為當前工作目錄。
--reload-include TEXT 監視檔案時包含檔案。預設包含 '*.py'
--reload-exclude TEXT 監視檔案時排除檔案。
--reload-delay FLOAT 上次檢查和下次檢查之間的延遲,預設為 0.25
--loop [auto|asyncio|uvloop] 事件迴圈實現。[預設值 auto]
--http [auto|h11|httptools] HTTP 協議實現。[預設值 auto]
--interface auto|asgi|wsgi 選擇應用程式介面。[預設值 auto]
--env-file PATH 環境配置檔案。
--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 測試工具

Falcon 是一個極簡主義框架,適用於開發 API。API 是兩個應用程式之間的介面。API 開發人員需要在將其釋出到生產環境之前測試其功能、可靠性、穩定性、可擴充套件性和效能等。

為此目的,可以使用各種 API 測試工具。在本節中,我們將學習如何使用命令列工具 **Curl** 和 **HTTPie**,以及一個名為 **Postman** 的 GUI 工具。

cURL

cURL 是一個開源專案,它提供 libcurl 庫和一個名為 curl 的命令列工具,該工具可以使用各種協議傳輸資料。支援 20 多種協議,包括 HTTP。縮寫 cURL 代表 Client URL。從命令列使用 Curl 的語法如下:

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

URL 引數由協議相關的一個或多個 URL 字串組成。Curl 命令可以使用各種選項進行自定義。一些重要的命令列選項如下:

  • **–X**: 指定請求方法。預設情況下,Curl 假設 GET 為請求方法。要傳送 POST、PUT 或 DELETE 請求,必須使用此選項。例如:

Curl –X DELETE https://:8000/student/1
  • **–H**: 此選項用於在請求中新增標頭。例如:

Curl –H "Content-Type: application/json" -X GET
https://:8000/students
  • **–i**: 當此選項包含在命令列中時,將顯示所有響應標頭。例如:

Curl –I –X DELETE https://:8000/student/2
  • **–d**: 要包含要在 HTTP 請求中進行處理的資料,我們必須使用此選項,尤其是在需要 POST 或 PUT 請求時。

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

HTTPie

HTTPie 是一個用 Python 編寫的命令列工具。據說它是一個“面向人類的類似 cURL 的工具”。它支援表單和檔案上傳,並生成格式良好的彩色終端輸出。與 Curl 相比,其表達性和直觀的語法使其更易於使用。

示例

  • **GET 請求** - http GET localhost:8000/students

  • **POST 請求** - http POST localhost:8000/students id=4 name="aaa" percent=50

  • **PUT 請求** - http PUT localhost:8000/students/2 id=3 name="Mathews" percent=55

  • **DELETE 請求** - http DELETE localhost:8000/students/2

Postman

Postman 是一款非常流行的 API 測試工具。它是一個 GUI 應用程式,而不是 Curl 和 HTTPie。它以瀏覽器外掛和桌面應用程式的形式提供。由於瀏覽器外掛不接受針對基於 localhost 的 API 的請求,因此我們需要從 https://www.postman.com/downloads. 下載桌面版本。

完成基於嚮導的安裝後,啟動 Postman 應用程式並建立一個新的請求。

Python Falcon API1

下拉選單顯示各種 HTTP 請求型別供選擇。

Python Falcon API2

在請求 URL 欄位中輸入 **https://:8000/hello**。右側的響應窗格顯示結果。

Python Falcon API3

當我們測試 Falcon API 對 SQLite 資料庫的 CRUD 操作時,稍後我們將使用相應的請求型別。

Python Falcon - 請求和響應

HTTP 協議規定客戶端向伺服器傳送 HTTP 請求,在伺服器上應用某些業務邏輯並制定響應,然後將響應重定向到客戶端。對於兩者之間的同步傳輸,Python 框架使用 WSGI 標準,而非同步傳輸遵循 ASGI 標準。Falcon 支援兩者。

WSGI/ASGI 伺服器在上下文資料中提供 Request 和 Response 物件。響應器、鉤子、中介軟體等使用這些物件作為引數。對於 WSGI 應用程式,處理 **falcon.Request** 類的例項。在 ASGI 應用程式中,它表示 **falcon.asgi.Request** 類。雖然不同,但這兩個類都被設計為具有相似的屬性和方法,以最大限度地減少混淆並允許更容易的移植性。

請求

Request 物件表示 HTTP 請求。由於它由伺服器提供,因此此物件並非旨在由響應器方法直接例項化。此物件提供以下屬性和方法,可在響應器、鉤子和中介軟體方法中使用:

  • **method** - 請求的 HTTP 方法(例如,“GET”、“POST”等)

  • **host** - Host 請求頭欄位

  • **port** - 用於請求的埠。返回給定模式的預設埠(HTTP 為 80,HTTPS 為 443)

  • **uri** - 請求的完全限定 URI。

  • **path** - 請求 URI 的路徑部分(不包括查詢字串)。

  • **query_string** - 請求 URI 的查詢字串部分,不包含前導 '?' 字元。

  • **cookies** - 名/值 Cookie 對的字典。

  • **content_type** - Content-Type 標頭的值,如果標頭缺失則為 None。

  • **stream** - 用於讀取請求正文(如果有的化)的檔案式輸入物件。此物件提供對伺服器資料流的直接訪問,並且不可搜尋。

  • **bounded_stream** - stream 的檔案式包裝器

  • **headers** - 來自請求的原始 HTTP 標頭

  • **params** - 請求查詢引數名稱與其值的對映。

  • **get_cookie_values(name)** - 返回 Cookie 標頭中為指定 Cookie 提供的所有值。cookies 屬性的別名。

  • **get_media()** - 返回請求流的反序列化形式。類似於 media 屬性。

  • **get_param(name)** - 將查詢字串引數的原始值作為字串返回。如果使用 **application/x-wwwform-urlencoded** 媒體型別釋出了 HTML 表單,則 Falcon 可以自動從請求正文解析引數並將它們合併到查詢字串引數中。要啟用此功能,請透過 **App.req_options** 將 **auto_parse_form_urlencoded** 設定為 True。

響應

Response 物件表示伺服器對客戶端的 HTTP 響應。與 Request 物件一樣,Response 物件也不應由響應器直接例項化。

響應器、鉤子函式或中介軟體方法透過訪問以下屬性和方法來操作此物件:

  • **status** - HTTP 狀態程式碼,例如“200 OK”。這可以設定為 **http.HTTPStatus** 的成員、HTTP 狀態行字串或位元組字串或整數。Falcon 提供了許多常用狀態程式碼的常量,以 HTTP_ 開頭,例如:**falcon.HTTP_204**。

  • **media** - 透過 **falcon.RequestOptions** 配置的媒體處理程式支援的可序列化物件。

  • **text** - 表示響應內容的字串。

  • **body** - text 的已棄用別名。

  • **data** - 表示響應內容的位元組字串。

  • **stream** - 表示響應內容的檔案式物件。

  • **content_length** - 設定 Content-Length 標頭。當未設定 text 或 data 屬性時,它會手動設定內容長度。

  • **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。

  • append_header (name, value) − 為此響應設定或追加標頭。用於設定cookie。

  • delete_header (name) − 刪除先前為此響應設定的標頭。

  • get_header (name) − 檢索給定標頭的原始字串值。

  • set_cookie (name, value) − 設定響應cookie。此方法可以多次呼叫以向響應新增一個或多個cookie。

  • set_header (name, value) − 將此響應的標頭設定為給定值。

  • set_stream (stream, content_length) − 設定流和內容長度。

  • unset_cookie (name, domain=None, path=None) − 在響應中取消設定cookie。此方法清除cookie的內容,並指示使用者代理立即過期其自己的cookie副本。

Python Falcon - 資源類

Falcon的設計借鑑了REST架構風格的幾個關鍵概念。REST代表表述性狀態轉移(Representational State Transfer)。REST定義了Web應用程式的架構應該如何執行。

REST是一種基於資源的架構。在這裡,REST伺服器託管的所有內容,無論是檔案、影像還是資料庫表中的行,都被視為資源,它可能具有多種表示形式。REST API提供對這些資源的受控訪問,以便客戶端可以檢索和修改它們。

伺服器上的資源應該只有一個統一資源識別符號 (URI)。它只標識資源;它不會指定對該資源採取什麼操作。相反,使用者可以選擇一組標準方法。HTTP動詞或方法用於對資源進行操作。POST、GET、PUT和DELETE方法分別執行CREATE、READ、UPDATE和DELETE操作。

Falcon使用普通的Python類來表示資源。這樣的類充當應用程式中的控制器。它將傳入的請求轉換為一個或多個內部操作,然後根據這些操作的結果向客戶端返回響應。

Python Falcon Resource

每個資源類都定義了各種“響應器”方法,每個資源允許的HTTP方法對應一個。響應器名稱以“on_”開頭,並根據它們處理的HTTP方法命名,例如on_get()、on_post()、on_put()等。

在上文中使用的hellofalcon.py示例程式碼中,HelloResource(資源類)有一個on_get()響應器方法。響應器必須始終定義至少兩個引數以接收Request和Response物件。

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物件表示傳入的HTTP請求。可以透過此物件訪問請求標頭、查詢字串引數以及與請求相關的其他元資料。

Response物件表示應用程式對請求的HTTP響應。此物件的屬性和方法設定狀態、標頭和正文資料。它還公開了一個類似字典的上下文屬性,用於將任意資料傳遞給鉤子和其他的中介軟體方法。

請注意,上述示例中的HelloResource只是一個普通的Python類。它可以有任意名稱;但是,約定是將其命名為xxxResource

Python Falcon - 應用類

此類是基於Falcon的WSGI應用程式的主要入口點。此類的例項提供可呼叫的WSGI介面和路由引擎。

import falcon
app = falcon.App()

此類的__init__()建構函式採用以下關鍵字引數:

  • media_type − 初始化RequestOptions和ResponseOptions時使用的媒體型別。Falcon允許輕鬆自定義internet媒體型別處理。預設情況下,Falcon僅啟用對JSON和HTML(URL編碼和多部分)表單的處理程式。

  • Falcon支援的其他媒體型別由以下常量表示:

    • falcon.MEDIA_JSON

    • falcon.MEDIA_MSGPACK

    • falcon.MEDIA_MULTIPART

    • falcon.MEDIA_URLENCODED

    • falcon.MEDIA_YAML

    • falcon.MEDIA_XML

    • falcon.MEDIA_HTML

    • falcon.MEDIA_JS

    • falcon.MEDIA_TEXT

    • falcon.MEDIA_JPEG

    • falcon.MEDIA_PNG

    • falcon.MEDIA_GIF

  • request_type − 此引數的預設值為falcon.Request類。

  • 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模板路由到資源。如果路徑與給定路由的模板匹配,則請求將傳遞給關聯的資源進行處理。根據請求方法,將呼叫相應的響應器方法。

示例

讓我們向HelloResource類新增on_post()響應器方法,並測試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)

輸出

使用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"""}" https://: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引數是要從中提供檔案的源目錄。如果要包含ContentDisposition標頭到響應中,則將downloadable引數設定為True。fallback_filename預設為None,但在找不到請求的檔案時可以指定。

add_error_handler()方法用於為一種或多種異常型別註冊處理程式。

add_error_handler(self, exception, handler=None)

WSGI 可呼叫 App 類擁有相同的方法。它在 falcon.asgi 模組中定義。

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

請注意, 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'
      )

Python Falcon - 路由

Falcon採用RESTful架構風格。因此它使用基於資源的路由。資源類負責透過響應器處理HTTP方法,響應器本質上是類方法,其名稱以on_開頭,以小寫的HTTP方法名稱結尾(例如,on_get()、on_patch()、on_delete()等)。Falcon Application物件的add_route()方法將其路由器與資源類的例項關聯。

在上文中使用的Hellofalcon.py示例中,當客戶端分別透過GET和POST方法請求/hello路由時,將呼叫on_get()on_post()響應器。

如果沒有路由與請求匹配,則會引發HTTPRouteNotFound例項。另一方面,如果路由匹配但資源未為請求的HTTP方法實現響應器,則預設響應器會引發HTTPMethodNotAllowed例項。

欄位轉換器

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工具測試https://: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的路由引擎具有以下內建欄位轉換器,可以使用它們將其讀取為其他資料型別。

  • IntConverter − 此類在falcon.routing模組中定義。建構函式使用以下引數:

IntConverter(num_digits=None, min=None, max=None)

    其中,

    • num_digits − 值必須具有給定的位數。

    • min − 引數的最小值

    • max − 引數的最大值。

    例如,以下add_route()函式接受1到100之間的整數作為rollno

app.add_route('/student/{rollno:int(1,1,100}', StudentResource())
  • UUIDConverter − falcon.routing模組中的此類將32個十六進位制數字的字串轉換為UUID(通用唯一識別符號)。

  • DateTimeConverter − 將引數字串轉換為datetime變數。引數必須是任何被strptime()函式識別的格式的字串,預設為'%Y-%m-%dT%H:%M:%SZ'

格式字串使用以下格式程式碼:

%a 縮寫的星期幾名稱 Sun, Mon
%A 完整的星期幾名稱 Sunday, Monday
%d 月份中的日期,為零填充的十進位制數 01, 02
%-d 月份中的日期,為十進位制數 1, 2..
%b 縮寫的月份名稱 Jan, Feb
%m 月份,為零填充的十進位制數 01, 02
%B 完整的月份名稱 January, February
%-y 年份(不含世紀),為十進位制數 0, 99
%Y 年份(含世紀),為十進位制數 2000, 1999
%H 小時(24小時制),為零填充的十進位制數 01, 23
%p 區域設定的AM或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檢查URLhttps://: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 - 字尾響應器

為了理解字尾響應器的概念和需求,讓我們定義一個StudentResource類。它包含一個on_get()響應器,該響應器將學生轉換為dict物件的列表,並將其作為響應返回。

讓我們還新增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的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'向Application物件新增帶有id引數的路由。

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()響應程式(刪除資源)。

我們已經添加了這個路由,字尾為student。因此,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()響應程式 simply 刪除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 模組

inspect模組是一個方便的工具,它提供有關已註冊路由以及Falcon應用程式的其他元件(如中介軟體、接收器等)的資訊。

應用程式的檢查可以透過兩種方式完成:CLI工具和程式設計方式。falcon-inspect工具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 模板

Falcon庫主要用於構建API和微服務。因此,預設情況下,Falcon響應程式返回JSON響應。但是,如果內容型別更改為falcon.MEDIA_HTML,則可以呈現HTML輸出。

使用可變資料呈現HTML內容非常繁瑣。為此,使用Web模板庫。許多Python Web框架都捆綁了特定的模板庫。但Falcon作為一個極簡的微型框架,並沒有預捆綁任何一個。

Jinja2是許多Python框架使用的最流行的模板庫之一。在本節中,我們將瞭解如何在Falcon應用程式中使用Jinja2。Jinja2是一個快速且對設計人員友好的模板語言,易於配置和除錯。它的沙箱環境使得防止執行不受信任的程式碼、禁止潛在的不安全資料以及防止跨站點指令碼攻擊(稱為XSS攻擊)變得容易。

Jinja2的另一個非常強大的功能是模板繼承,您可以定義一個具有公共設計特徵的基模板,子模板可以覆蓋這些特徵。

首先,使用PIP實用程式在當前Python環境中安裝Jinja2

pip3 install jinja2

Hello World 模板

Jinja2模組定義了一個Template類。透過讀取包含HTML指令碼的檔案(副檔名為.html的檔案)的內容來獲得Template物件。透過呼叫此Template物件的render()方法,可以將HTML響應呈現給客戶端瀏覽器。Response物件的content_type屬性必須設定為falcon.MEDIA_HTML

讓我們將以下HTML指令碼儲存為應用程式資料夾中的hello.py

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

示例

下面的資源類中的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)

輸出

執行上述Python程式碼,並在瀏覽器中訪問https://:8000/hello連結。

Jinja2

模板變數

Jinja2是一個伺服器端模板庫。網頁透過將Jinja2模板語言的各種元素作為佔位符放在HTML指令碼內的適當分隔符內來構建為模板。模板引擎讀取HTML指令碼,用伺服器上的上下文資料替換佔位符,重新組合HTML,並將其呈現給客戶端。

Template.render()函式有一個可選的上下文字典引數。此字典的關鍵屬性成為模板變數。這有助於在網頁中呈現響應程式傳遞的資料。

示例

在下面的示例中,路由/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在一個名為name的模板變數中讀取路徑引數。它在HTML指令碼中充當佔位符。它放在{{}}符號中,以便其值顯示為HTML響應。

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

輸出

執行Python程式碼並輸入https://:8000/hello/Priya作為URL。瀏覽器將顯示以下輸出:

Jinja2 Hello

Jinja2模板中的迴圈

如果響應程式傳遞任何Python可迭代物件,例如列表、元組或字典,則可以使用其迴圈結構語法在Jinja2模板內遍歷其元素。

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

在下面的示例中,on_get()響應程式將students物件(一個dict物件的列表)傳送到模板list.html。它依次遍歷資料並將其呈現為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 Image

HTML表單模板

在本節中,我們將瞭解Falcon如何讀取HTML表單中的資料。讓我們將以下HTML指令碼儲存為myform.html。我們將使用它來獲取模板物件並呈現它。

<html>
<body>
   <form method="POST" action="https://: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路由。

為了能夠讀取表單資料,必須將falcon.RequestOptions類的auto_parse_form_urlencoded屬性設定為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表單。

Jinja2 Host

students資料庫字典將被附加。訪問/students路由。您將看到一個新的行被附加。

Jinja2 Example

多部分表單

為了讓使用者從本地檔案系統選擇檔案,HTML表單的enctype屬性必須設定為multipart/form-data。Falcon使用MultipartFormHandler處理multipart/form-data媒體型別,允許它迭代表單中的主體部分。

BodyPart類具有以下屬性:

  • stream - 僅適用於當前主體部分的流包裝器

  • data - 主體部分內容位元組

  • content_type 如果未指定,則預設為text/plain,符合RFC

  • text - 當前主體部分解碼為文字字串(僅當型別為text/plain時提供,否則為None)

  • media - 透過媒體處理程式自動解析,方式與req.media相同

  • name, filename - 來自Content-Disposition標頭的相關部分

  • secure_filename - 經過清理的檔名,可以在伺服器檔案系統安全使用。

以下HTML指令碼(index.html)是一個多部分表單。

<html>
   <body>
      <form action="https://: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>

此表單由以下程式碼中HelloResource類的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)

執行上述程式並訪問https://: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 - Cookie

cookie以文字檔案的形式儲存在客戶端的計算機上。其目的是記住和跟蹤與客戶端使用相關的 資料,以獲得更好的訪問者體驗和網站統計資訊。

Request物件包含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物件具有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.Request 實現與 falcon.Request 相同的 cookie 方法和屬性。 ASGI 版本的 set_cookie()append_header() 是同步的,因此不需要等待它們。

Python Falcon - 狀態碼

預設情況下,HTTP伺服器對客戶端請求的響應具有200 OK狀態。Falcon提供了自己的狀態常量列表,以提高方便性和可讀性。

例如,200 OK狀態碼錶示為:

resp.status = falcon.HTTP_OK

這些預定義的Falcon常量避免了拼寫錯誤,並減少了準備響應時必須建立的字串物件數量。但是,從Falcon 3.0版本開始,也允許使用裸int程式碼。

resp.status = 200

對於 ASGI 應用程式,相同的狀態程式碼適用。

以下是Falcon庫中定義的一些狀態程式碼:

資訊程式碼

  • HTTP_CONTINUE = HTTP_100

  • HTTP_SWITCHING_PROTOCOLS = HTTP_101

  • HTTP_PROCESSING = HTTP_102

成功狀態程式碼

  • HTTP_OK = HTTP_200

  • HTTP_CREATED = HTTP_201

  • HTTP_ACCEPTED = HTTP_202

  • HTTP_NON_AUTHORITATIVE_INFORMATION = HTTP_203

  • HTTP_NO_CONTENT = HTTP_204

  • HTTP_RESET_CONTENT = HTTP_205

  • HTTP_PARTIAL_CONTENT = HTTP_206

  • HTTP_MULTI_STATUS = HTTP_207

  • HTTP_ALREADY_REPORTED = HTTP_208

  • HTTP_IM_USED = HTTP_226

重定向錯誤程式碼

  • HTTP_MULTIPLE_CHOICES = HTTP_300

  • HTTP_MOVED_PERMANENTLY = HTTP_301

  • HTTP_FOUND = HTTP_302

  • HTTP_SEE_OTHER = HTTP_303

  • HTTP_NOT_MODIFIED = HTTP_304

  • HTTP_USE_PROXY = HTTP_305

  • HTTP_TEMPORARY_REDIRECT = HTTP_307

  • HTTP_PERMANENT_REDIRECT = HTTP_308

客戶端錯誤程式碼

  • HTTP_BAD_REQUEST = HTTP_400

  • HTTP_UNAUTHORIZED = HTTP_401 # "未經授權"

  • HTTP_PAYMENT_REQUIRED = HTTP_402

  • HTTP_FORBIDDEN = HTTP_403 # "未授權"

  • HTTP_NOT_FOUND = HTTP_404

  • HTTP_METHOD_NOT_ALLOWED = HTTP_405

  • HTTP_NOT_ACCEPTABLE = HTTP_406

  • HTTP_PROXY_AUTHENTICATION_REQUIRED = HTTP_407

  • HTTP_REQUEST_TIMEOUT = HTTP_408

  • HTTP_CONFLICT = HTTP_409

伺服器錯誤程式碼

  • HTTP_INTERNAL_SERVER_ERROR = HTTP_500

  • HTTP_NOT_IMPLEMENTED = HTTP_501

  • HTTP_BAD_GATEWAY = HTTP_502

  • HTTP_SERVICE_UNAVAILABLE = HTTP_503

  • HTTP_GATEWAY_TIMEOUT = HTTP_504

  • HTTP_HTTP_VERSION_NOT_SUPPORTED = HTTP_505

  • HTTP_INSUFFICIENT_STORAGE = HTTP_507

  • HTTP_LOOP_DETECTED = HTTP_508

  • HTTP_NETWORK_AUTHENTICATION_REQUIRED = HTTP_511

Python Falcon - 錯誤處理

為了處理各種錯誤情況,可以使用上述狀態程式碼來處理響應物件。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"
      )

預定義錯誤類

Falcon提供的一些預定義錯誤類如下:

  • **HTTPBadRequest** − 400 錯誤請求。由於客戶端錯誤(例如請求語法錯誤、無效的請求訊息框架等),伺服器無法處理請求。

  • **HTTPInvalidHeader** − 導致400錯誤請求,因為請求中的一個或多個頭部無效。

  • **HTTPInvalidParam** − 表示400錯誤請求。此錯誤可能指提交給請求的查詢字串、表單或文件中的無效引數。

  • **HTTPMissingParam** − 請求中缺少引數時引發400錯誤請求。

  • **HTTPForbidden** − 伺服器理解了請求,但拒絕授權。狀態碼為403禁止。

  • **HTTPNotFound** − 當伺服器找不到目標資源的當前表示時,會引發404狀態碼。它不指示這種表示的缺乏是暫時的還是永久的。

  • **HTTPMethodNotAllowed** − 405 方法不允許。請求行中收到的方法不受目標資源支援。

  • **HTTPLengthRequired** − 當伺服器拒絕在沒有定義Content-Length的情況下接受請求時。411 需要長度。錯誤程式碼。

  • **HTTPUnsupportedMediaType** − 如果源伺服器由於有效負載採用此方法在目標資源上不支援的格式而拒絕服務請求。等效狀態碼為415不支援的媒體型別。

  • **HTTPUnprocessableEntity** − 如果伺服器理解請求實體的內容型別並且請求實體的語法正確,但無法處理所包含的指令,則引發的錯誤狀態碼為422無法處理的實體。例如,如果XML請求正文包含格式良好但語義錯誤的XML指令。

  • **HTTPTooManyRequests** − 當用戶在給定時間內傳送了過多的請求(“速率限制”)時,會引發429過多請求狀態碼。

  • **HTTPInternalServerError** − 一個非常常見的錯誤情況,導致500內部伺服器錯誤。伺服器遇到意外情況,阻止其完成請求。

  • **HTTPNotImplemented** − 501(未實現)狀態碼錶示伺服器不支援完成請求所需的功能。當伺服器無法識別請求方法並且無法為任何資源支援它時,這是適當的響應。

  • **HTTPServiceUnavailable** − 503服務不可用表示伺服器目前由於臨時過載或計劃維護而無法處理請求。

  • **MediaNotFoundError** − 400錯誤請求。媒體處理程式在嘗試解析空正文時引發此異常。

  • **MediaMalformedError** − 400錯誤請求。媒體處理程式在嘗試解析格式錯誤的正文時引發此異常。

重定向

還有一些異常,當引發這些異常時,會觸發對客戶端的重定向響應。狀態碼為3xx型別。這些異常由以下類表示,作為**HttpError**的子類,會中斷請求處理。

  • **HTTPMovedPermanently** − 301永久移動。此狀態碼錶示目標資源已分配新的永久URI。

  • **HTTPFound** − 302已找到狀態碼,表示目標資源暫時位於不同的URI下。

  • **HTTPTemporaryRedirect** − 此類引發307(臨時重定向)狀態碼,這意味著目標資源暫時位於不同的URI下,並且如果使用者代理對此URI執行自動重定向,則使用者代理必須不更改請求方法。

  • **HTTPPermanentRedirect** − 導致308永久重定向,表示目標資源已分配新的永久URI。

Python Falcon - Hook

鉤子是使用者定義的函式,在響應客戶端請求時呼叫資源類中的特定響應程式方法時會自動執行。Falcon支援**before**和**after**鉤子。

要作為鉤子使用的函式,除了任何必要的可選引數外,還使用請求、響應和資源類作為引數進行定義。

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

透過應用以下裝飾器之一,可以將此類函式附加到單個響應程式或整個資源類:

  • @falcon.before(hookfunction)

  • @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 - 中介軟體

**“中介軟體”**是在處理每個請求(在任何特定響應程式處理之前)以及在返回每個響應之前處理的函式。此函式接收進入應用程式的每個請求。

中介軟體的工作方式類似於鉤子。但是,與鉤子不同,中介軟體方法全域性應用於整個應用程式。它可以透過執行其中定義的程式碼來對請求執行某些處理,然後將請求傳遞給相應的操作函式進行處理。它還可以在返回生成的響應之前處理操作函式生成的響應。

中介軟體是一個實現以下一個或多個事件處理程式方法的類。對於WSGI應用程式,這些方法是:

  • **process_request (self, req, resp)** − 此方法在路由請求之前處理請求。

  • **process_resource (self, req, resp, resource, params)** − 路由後處理請求。可以傳遞一個**dict**物件,該物件表示從路由的URI模板欄位派生的任何附加引數。

  • **process_response (self, req, resp, resource, req_succeeded)** − 此方法用於響應的後處理(路由後)。如果未引發異常,則**req_succeeded**引數為True,否則為False。

對於 ASGI 應用程式,除了上述方法外,中介軟體類還可以定義更多方法。

為了考慮生命週期事件(WSGI規範的一個可選部分),可以包含啟動和關閉事件處理程式。

  • **process_startup (self, scope, event)** − 此方法處理 ASGI 生命週期啟動事件。當伺服器準備啟動並接收連線但尚未開始這樣做時,就會呼叫它。

  • **process_shutdown(self, scope, event)** − 此方法處理 ASGI 生命週期關閉事件。當伺服器停止接受連線並關閉所有活動連線時,就會呼叫它。

由於 ASGI 應用程式也響應 Websocket 協議下的請求,因此中介軟體可以定義以下協程方法:

  • **process_request_ws (self, req, ws)** − 此方法在路由 WebSocket 握手請求之前處理它。

  • **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

**“跨源資源共享”(CORS)**是一種情況,其中在一個客戶端瀏覽器上執行的前端應用程式嘗試透過 JavaScript 程式碼與後端通訊,而後端位於與前端不同的“源”中。此處的源是協議、域名和埠號的組合。因此,**https://** 和 **https://** 具有不同的來源。

如果具有一個源 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** 是客戶端和伺服器之間持久的連線,用於在兩者之間提供雙向、**全雙工**通訊。通訊透過單個 TCP/IP 套接字連線透過 HTTP 進行。它可以被視為 HTTP 的升級,而不是協議本身。

HTTP 的一個限制是它是一種嚴格的半雙工或單向協議。另一方面,使用 WebSockets,我們可以傳送基於訊息的資料,類似於 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 物件。

Falcon 中 WebSocket 的工作原理?

以下示例演示了 Falcon 應用程式中 WebSocket 的功能。首先,我們有一個渲染模板的**on_get()**響應程式。

示例

客戶端瀏覽器顯示一個帶有文字欄位和按鈕的表單,單擊按鈕時,將建立**websocket**物件,並觸發**on_websocket()**響應程式。它接受使用者輸入的訊息並將其回顯到客戶端,字首為“The message text was”。

import falcon
import falcon.asgi
import jinja2
html = """
<!DOCTYPE html>
<html>
   <head>
      <title>Chat</title>
   </head>
   <body>
      <script>
         var ws = new WebSocket("ws://: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)

輸出

啟動 Uvicorn 伺服器並訪問**https://:8000/ws** URL 以顯示聊天表單。

Websocket Image

輸入一些文字並按**傳送**按鈕。

Websocket Example

Python Falcon - SQLAlchemy 模型

為了演示 Falcon 的響應程式函式(**on_post()、on_get()、on_put()** 和 **on_delete()**)的工作方式,我們對記憶體資料庫(以 Python 字典物件列表的形式)執行了**CRUD**(代表建立、檢索、更新和刪除)操作。相反,我們可以使用任何關係資料庫(例如 MySQL、Oracle 等)來執行儲存、檢索、更新和刪除操作。

我們不使用符合**DB-API**的資料庫驅動程式,而是使用**SQLAlchemy**作為 Python 程式碼和資料庫之間的介面(我們將使用 SQLite 資料庫,因為 Python 對其有內建支援)。SQLAlchemy 是一個流行的 SQL 工具包和**物件關係對映器**。

物件關係對映是一種程式設計技術,用於在面向物件的程式語言中轉換不相容型別系統之間的資料。通常,在像 Python 這樣的面嚮物件語言中使用的型別系統包含非標量型別。但是,大多數資料庫產品(如 Oracle、MySQL 等)中的資料型別是原始型別,例如整數和字串。

在ORM系統中,每個類都對映到底層資料庫中的一個表。ORM替你處理了這些繁瑣的資料庫互動程式碼,讓你可以專注於系統邏輯的程式設計。

為了使用SQLAlchemy,我們需要首先使用PIP安裝器安裝該庫。

pip install sqlalchemy

SQLAlchemy設計用於與為特定資料庫構建的DBAPI實現一起工作。它使用方言系統與各種型別的DBAPI實現和資料庫進行通訊。所有方言都需要安裝相應的DBAPI驅動程式。

包含的方言如下:

  • Firebird

  • Microsoft SQL Server

  • MySQL

  • Oracle

  • PostgreSQL

  • SQLite

  • Sybase

資料庫引擎

由於我們將使用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})

為了與資料庫互動,我們需要獲取它的控制代碼。會話物件是資料庫的控制代碼。Session類使用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()

模型類

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()方法在資料庫中建立相應的表。可以使用SQLite視覺化工具(如SQLiteStudio)進行確認。

Sqlite

現在,我們需要宣告一個StudentResource類,其中定義了HTTP響應方法,用於對students表執行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"和marks=760)作為主體引數傳遞。請求成功處理,並將一行新增到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響應,其中顯示students模型中所有物件的資料。

Postman Example

Postman應用程式結果窗格中顯示的兩條記錄也可以在SQLiteStudio的資料檢視中進行驗證。

Python Sqlite1

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=2的物件,並更改名稱和分數。請注意,這些值作為主體引數傳遞。

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 - 測試

Falcon的測試模組是Falcon應用程式的功能測試框架。它包含各種測試類和實用程式函式,以支援功能測試。該測試框架同時支援unittestpytest

我們將使用以下指令碼(myapp.py)來演示測試功能。它包含一個HelloResource類,該類具有一個on_get()響應程式,該響應程式呈現Hello World的JSON響應。create()函式返回添加了已註冊'/' URL路由的Falcon應用程式物件。

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)

使用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)

使用Pytest

要使用PyTest框架執行測試,需要使用PIP實用程式安裝它。

pip3 install pytest

要執行test函式,我們需要一個testing.TestClient類的物件。它模擬WSGI和ASGI應用程式的請求。首先透過將Falcon應用程式物件作為引數來獲取此物件。

我們執行simulate_*()函式,並將其結果與預期輸出進行斷言,以確定測試是否失敗或透過。在這兩個示例中,測試都由於“Hello World”訊息中“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 - 部署

可以使用啟用了mod_wsgi模組的Apache伺服器來部署Falcon Web應用程式,就像任何WSGI應用程式一樣。另一種替代方案是使用uWSGIgunicorn進行部署。

uWSGI是一個快速且高度可配置的WSGI伺服器。如果與NGINX一起使用,它可以在生產就緒環境中以速度的形式提供更好的效能。

首先,使用PIP安裝程式在Python虛擬環境中安裝Falcon和uWSGI,並使用wsgi.py將其公開給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配置檔案,如下所示:

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伺服器。您應該有一個正在執行的應用程式。

廣告
© . All rights reserved.