從零構建一個簡單的 Python Web框架

時間 2019-11-30

標籤構建一個簡單 python web 框架欄目 Python 简体版

原文原文鏈接

爲何你想要本身構建一個 web 框架呢？我想，緣由有如下幾點：html

你有一個新奇的想法，以爲將會取代其餘的框架
你想要得到一些名氣
你遇到的問題很獨特，以致於現有的框架不太合適
你對 web 框架是如何工做的很感興趣，由於你想要成爲一位更好的 web 開發者。

接下來的筆墨將着重於最後一點。這篇文章旨在經過對設計和實現過程一步一步的闡述告訴讀者，我在完成一個小型的服務器和框架以後學到了什麼。你能夠在這個代碼倉庫中找到這個項目的完整代碼。python

我但願這篇文章能夠鼓勵更多的人來嘗試，由於這確實頗有趣。它讓我知道了 web 應用是如何工做的，並且這比我想的要容易的多！linux

範圍

框架能夠處理請求-響應週期、身份認證、數據庫訪問、模板生成等部分工做。Web 開發者使用框架是由於，大多數的 web 應用擁有大量相同的功能，而對每一個項目都從新實現一樣的功能意義不大。git

比較大的的框架如 Rails 和 Django 實現了高層次的抽象，或者說「自備電池」（「batteries-included」，這是 Python 的口號之一，意即全部功能都自足。）。而實現全部的這些功能可能要花費數千小時，所以在這個項目上，咱們重點完成其中的一小部分。在開始寫代碼前，我先列舉一下所需的功能以及限制。github

功能：web

處理 HTTP 的 GET 和 POST 請求。你能夠在這篇 wiki 中對 HTTP 有個大體的瞭解。
實現異步操做（我喜歡 Python 3 的 asyncio 模塊）。
簡單的路由邏輯以及參數擷取。
像其餘微型框架同樣，提供一個簡單的用戶級 API 。
支持身份認證，由於學會這個很酷啊（微笑）。

限制：正則表達式

將只支持 HTTP 1.1 的一個小子集，不支持傳輸編碼transfer-encoding、HTTP 認證http-auth、內容編碼content-encoding（如 gzip）以及持久化鏈接等功能。
不支持對響應內容的 MIME 判斷 - 用戶須要手動指定。
不支持 WSGI - 僅能處理簡單的 TCP 鏈接。
不支持數據庫。

我以爲一個小的用例可讓上述內容更加具體，也能夠用來演示這個框架的 API：數據庫

from diy_framework import App, Router
from diy_framework.http_utils import Response
# GET simple route
async def home(r):
rsp = Response()
rsp.set_header('Content-Type', 'text/html')
rsp.body = '<html><body><b>test</b></body></html>'
return rsp
# GET route + params
async def welcome(r, name):
return "Welcome {}".format(name)
# POST route + body param
async def parse_form(r):
if r.method == 'GET':
return 'form'
else:
name = r.body.get('name', '')[0]
password = r.body.get('password', '')[0]
return "{0}:{1}".format(name, password)
# application = router + http server
router = Router()
router.add_routes({
r'/welcome/{name}': welcome,
r'/': home,
r'/login': parse_form,})
app = App(router)
app.start_server()

' 用戶須要定義一些可以返回字符串或 Response 對象的異步函數，而後將這些函數與表示路由的字符串配對，最後經過一個函數調用（start_server）開始處理請求。express

完成設計以後，我將它抽象爲幾個我須要編碼的部分：數組

接受 TCP 鏈接以及調度一個異步函數來處理這些鏈接的部分
將原始文本解析成某種抽象容器的部分
對於每一個請求，用來決定調用哪一個函數的部分
將上述部分集中到一塊兒，併爲開發者提供一個簡單接口的部分

我先編寫一些測試，這些測試被用來描述每一個部分的功能。幾回重構後，整個設計被分紅若干部分，每一個部分之間是相對解耦的。這樣就很是好，由於每一個部分能夠被獨立地研究學習。如下是我上文列出的抽象的具體體現：

一個 HTTPServer 對象，須要一個 Router 對象和一個 http_parser 模塊，並使用它們來初始化。
HTTPConnection 對象，每個對象表示一個單獨的客戶端 HTTP 鏈接，而且處理其請求-響應週期：使用 http_parser 模塊將收到的字節流解析爲一個 Request 對象；使用一個 Router 實例尋找並調用正確的函數來生成一個響應；最後將這個響應發送回客戶端。
一對 Request 和 Response 對象爲用戶提供了一種友好的方式，來處理實質上是字節流的字符串。用戶不須要知道正確的消息格式和分隔符是怎樣的。
一個包含「路由:函數」對應關係的 Router 對象。它提供一個添加配對的方法，能夠根據 URL 路徑查找到相應的函數。
最後，一個 App 對象。它包含配置信息，並使用它們實例化一個 HTTPServer 實例。

讓咱們從 HTTPConnection 開始來說解各個部分。

模擬異步鏈接

爲了知足上述約束條件，每個 HTTP 請求都是一個單獨的 TCP 鏈接。這使得處理請求的速度變慢了，由於創建多個 TCP 鏈接須要相對高的花銷（DNS 查詢，TCP 三次握手，慢啓動等等的花銷），不過這樣更加容易模擬。對於這一任務，我選擇相對高級的 asyncio-stream 模塊，它創建在 asyncio 的傳輸和協議的基礎之上。我強烈推薦你讀一讀標準庫中的相應代碼，頗有意思！

一個 HTTPConnection 的實例可以處理多個任務。首先，它使用 asyncio.StreamReader 對象以增量的方式從 TCP 鏈接中讀取數據，並存儲在緩存中。每個讀取操做完成後，它會嘗試解析緩存中的數據，並生成一個 Request 對象。一旦收到了這個完整的請求，它就生成一個回覆，並經過 asyncio.StreamWriter 對象發送回客戶端。固然，它還有兩個任務：超時鏈接以及錯誤處理。

你能夠在這裏瀏覽這個類的完整代碼。我將分別介紹代碼的每一部分。爲了簡單起見，我移除了代碼文檔。

class HTTPConnection(object):
def init(self, http_server, reader, writer):
self.router = http_server.router
self.http_parser = http_server.http_parser
self.loop = http_server.loop
self._reader = reader
self._writer = writer
self._buffer = bytearray()
self._conn_timeout = None
self.request = Request()

這個 init 方法沒啥意思，它僅僅是收集了一些對象以供後面使用。它存儲了一個 router 對象、一個 http_parser 對象以及 loop 對象，分別用來生成響應、解析請求以及在事件循環中調度任務。

而後，它存儲了表明一個 TCP 鏈接的讀寫對，和一個充當原始字節緩衝區的空字節數組。_conn_timeout 存儲了一個 asyncio.Handle 的實例，用來管理超時邏輯。最後，它還存儲了 Request 對象的一個單一實例。

下面的代碼是用來接受和發送數據的核心功能：

async def handle_request(self):
try:
while not self.request.finished and not self._reader.at_eof():
data = await self._reader.read(1024)
if data:
self._reset_conn_timeout()
await self.process_data(data)
if self.request.finished:
await self.reply()
elif self._reader.at_eof():
raise BadRequestException()
except (NotFoundException,
BadRequestException) as e:
self.error_reply(e.code, body=Response.reason_phrases[e.code])
except Exception as e:
self.error_reply(500, body=Response.reason_phrases[500])
self.close_connection()

全部內容被包含在 try-except 代碼塊中，這樣在解析請求或響應期間拋出的異常能夠被捕獲到，而後一個錯誤響應會發送回客戶端。

在 while 循環中不斷讀取請求，直到解析器將 self.request.finished 設置爲 True ，或者客戶端關閉鏈接所觸發的信號使得 self._reader_at_eof() 函數返回值爲 True 爲止。這段代碼嘗試在每次循環迭代中從 StreamReader 中讀取數據，並經過調用 self.process_data(data) 函數以增量方式生成 self.request。每次循環讀取數據時，鏈接超時計數器被重置。

這兒有個錯誤，你發現了嗎？稍後咱們會再討論這個。須要注意的是，這個循環可能會耗盡 CPU 資源，由於若是沒有讀取到東西 self._reader.read() 函數將會返回一個空的字節對象 b''。這就意味着循環將會不斷運行，卻什麼也不作。一個可能的解決方法是，用非阻塞的方式等待一小段時間：await asyncio.sleep(0.1)。咱們暫且不對它作優化。

還記得上一段我提到的那個錯誤嗎？只有從 StreamReader 讀取數據時，self._reset_conn_timeout() 函數纔會被調用。這就意味着，直到第一個字節到達時，timeout 才被初始化。若是有一個客戶端創建了與服務器的鏈接卻不發送任何數據，那就永遠不會超時。這可能被用來消耗系統資源，從而致使拒絕服務式攻擊（DoS）。修復方法就是在 init 函數中調用 self._reset_conn_timeout() 函數。

當請求接受完成或鏈接中斷時，程序將運行到 if-else 代碼塊。這部分代碼會判斷解析器收到完整的數據後是否完成了解析。若是是，好，生成一個回覆併發送回客戶端。若是不是，那麼請求信息可能有錯誤，拋出一個異常！最後，咱們調用 self.close_connection 執行清理工做。

解析請求的部分在 self.process_data 方法中。這個方法很是簡短，也易於測試：

async def process_data(self, data):
self._buffer.extend(data)
self._buffer = self.http_parser.parse_into(
self.request, self._buffer)

每一次調用都將數據累積到 self._buffer 中，而後試着用 self.http_parser 來解析已經收集的數據。這裏須要指出的是，這段代碼展現了一種稱爲依賴注入（Dependency Injection）的模式。若是你還記得 init 函數的話，應該知道咱們傳入了一個包含 http_parser 對象的 http_server 對象。在這個例子裏，http_parser 對象是 diy_framework 包中的一個模塊。不過它也能夠是任何含有 parse_into 函數的類，這個 parse_into 函數接受一個 Request 對象以及字節數組做爲參數。這頗有用，緣由有二：一是，這意味着這段代碼更易擴展。若是有人想經過一個不一樣的解析器來使用 HTTPConnection，沒問題，只需將它做爲參數傳入便可。二是，這使得測試更加容易，由於 http_parser 不是硬編碼的，因此使用虛假數據或者 mock對象來替代是很容易的。

下一段有趣的部分就是 reply 方法了：

async def reply(self):
request = self.request
handler = self.router.get_handler(request.path)
response = await handler.handle(request)
if not isinstance(response, Response):
response = Response(code=200, body=response)
self._writer.write(response.to_bytes())
await self._writer.drain()

這裏，一個 HTTPConnection 的實例使用了 HTTPServer 中的 router 對象來獲得一個生成響應的對象。一個路由能夠是任何一個擁有 get_handler 方法的對象，這個方法接收一個字符串做爲參數，返回一個可調用的對象或者拋出 NotFoundException 異常。而這個可調用的對象被用來處理請求以及生成響應。處理程序由框架的使用者編寫，如上文所說的那樣，應該返回字符串或者 Response 對象。Response 對象提供了一個友好的接口，所以這個簡單的 if 語句保證了不管處理程序返回什麼，代碼最終都獲得一個統一的 Response 對象。

接下來，被賦值給 self._writer 的 StreamWriter 實例被調用，將字節字符串發送回客戶端。函數返回前，程序在 await self._writer.drain() 處等待，以確保全部的數據被髮送給客戶端。只要緩存中還有未發送的數據，self._writer.close() 方法就不會執行。

HTTPConnection 類還有兩個更加有趣的部分：一個用於關閉鏈接的方法，以及一組用來處理超時機制的方法。首先，關閉一條鏈接由下面這個小函數完成：

def close_connection(self):
self._cancel_conn_timeout()
self._writer.close()

每當一條鏈接將被關閉時，這段代碼首先取消超時，而後把鏈接從事件循環中清除。

超時機制由三個相關的函數組成：第一個函數在超時後給客戶端發送錯誤消息並關閉鏈接；第二個函數用於取消當前的超時；第三個函數調度超時功能。前兩個函數比較簡單，我將詳細解釋第三個函數 _reset_cpmm_timeout() 。

def _conn_timeout_close(self):
self.error_reply(500, 'timeout')
self.close_connection()
def _cancel_conn_timeout(self):
if self._conn_timeout:
self._conn_timeout.cancel()
def _reset_conn_timeout(self, timeout=TIMEOUT):
self._cancel_conn_timeout()
self._conn_timeout = self.loop.call_later(
timeout, self._conn_timeout_close)

每當 _reset_conn_timeout 函數被調用時，它會先取消以前全部賦值給 self._conn_timeout 的 asyncio.Handle 對象。而後，使用 BaseEventLoop.call_later 函數讓 _conn_timeout_close 函數在超時數秒（timeout）後執行。若是你還記得 handle_request 函數的內容，就知道每當接收到數據時，這個函數就會被調用。這就取消了當前的超時而且從新安排 _conn_timeout_close 函數在超時數秒（timeout）後執行。只要接收到數據，這個循環就會不斷地重置超時回調。若是在超時時間內沒有接收到數據，最後函數 _conn_timeout_close 就會被調用。

建立鏈接

咱們須要建立 HTTPConnection 對象，而且正確地使用它們。這一任務由 HTTPServer 類完成。HTTPServer類是一個簡單的容器，能夠存儲着一些配置信息（解析器，路由和事件循環實例），並使用這些配置來建立 HTTPConnection 實例：

class HTTPServer(object):
def init(self, router, http_parser, loop):
self.router = router
self.http_parser = http_parser
self.loop = loop
async def handle_connection(self, reader, writer):
connection = HTTPConnection(self, reader, writer)
asyncio.ensure_future(connection.handle_request(), loop=self.loop)

HTTPServer 的每個實例可以監聽一個端口。它有一個 handle_connection 的異步方法來建立 HTTPConnection 的實例，並安排它們在事件循環中運行。這個方法被傳遞給 asyncio.start_server 做爲一個回調函數。也就是說，每當一個 TCP 鏈接初始化時（以 StreamReader 和 StreamWriter 爲參數），它就會被調用。

self._server = HTTPServer(self.router, self.http_parser, self.loop)
self._connection_handler = asyncio.start_server(
self._server.handle_connection,
host=self.host,
port=self.port,
reuse_address=True,
reuse_port=True,
loop=self.loop)

這就是構成整個應用程序工做原理的核心：asyncio.start_server 接受 TCP 鏈接，而後在一個預配置的 HTTPServer 對象上調用一個方法。這個方法將處理一條 TCP 鏈接的全部邏輯：讀取、解析、生成響應併發送回客戶端、以及關閉鏈接。它的重點是 IO 邏輯、解析和生成響應。

講解了核心的 IO 部分，讓咱們繼續。

解析請求

這個微型框架的使用者被寵壞了，不肯意和字節打交道。它們想要一個更高層次的抽象 —— 一種更加簡單的方法來處理請求。這個微型框架就包含了一個簡單的 HTTP 解析器，可以將字節流轉化爲 Request 對象。

這些 Request 對象是像這樣的容器：

class Request(object):
def init(self):
self.method = None
self.path = None
self.query_params = {}
self.path_params = {}
self.headers = {}
self.body = None
self.body_raw = None
self.finished = False

它包含了全部須要的數據，能夠用一種容易理解的方法從客戶端接受數據。哦，不包括 cookie ，它對身份認證是很是重要的，我會將它留在第二部分。

每個 HTTP 請求都包含了一些必需的內容，如請求路徑和請求方法。它們也包含了一些可選的內容，如請求體、請求頭，或是 URL 參數。隨着 REST 的流行，除了 URL 參數，URL 自己會包含一些信息。好比，"/user/1/edit" 包含了用戶的 id 。

一個請求的每一個部分都必須被識別、解析，並正確地賦值給 Request 對象的對應屬性。HTTP/1.1 是一個文本協議，事實上這簡化了不少東西。（HTTP/2 是一個二進制協議，這又是另外一種樂趣了）

解析器不須要跟蹤狀態，所以 http_parser 模塊其實就是一組函數。調用函數須要用到 Request 對象，並將它連同一個包含原始請求信息的字節數組傳遞給 parse_into 函數。而後解析器會修改 Request 對象以及充當緩存的字節數組。字節數組的信息被逐漸地解析到 request 對象中。

http_parser 模塊的核心功能就是下面這個 parse_into 函數：

def parse_into(request, buffer):
_buffer = buffer[:]
if not request.method and can_parse_request_line(_buffer):
(request.method, request.path,
request.query_params) = parse_request_line(_buffer)
remove_request_line(_buffer)
if not request.headers and can_parse_headers(_buffer):
request.headers = parse_headers(_buffer)
if not has_body(request.headers):
request.finished = True
remove_intro(_buffer)
if not request.finished and can_parse_body(request.headers, _buffer):
request.body_raw, request.body = parse_body(request.headers, _buffer)
clear_buffer(_buffer)
request.finished = True
return _buffer

從上面的代碼中能夠看到，我把解析的過程分爲三個部分：解析請求行（這行像這樣：GET /resource HTTP/1.1），解析請求頭以及解析請求體。

請求行包含了 HTTP 請求方法以及 URL 地址。而 URL 地址則包含了更多的信息：路徑、url 參數和開發者自定義的 url 參數。解析請求方法和 URL 仍是很容易的 - 合適地分割字符串就行了。函數 urlparse.parse 能夠用來解析 URL 參數。開發者自定義的 URL 參數能夠經過正則表達式來解析。

接下來是 HTTP 頭部。它們是一行行由鍵值對組成的簡單文本。問題在於，可能有多個 HTTP 頭有相同的名字，卻有不一樣的值。一個值得關注的 HTTP 頭部是 Content-Length，它描述了請求體的字節長度（不是整個請求，僅僅是請求體）。這對於決定是否解析請求體有很重要的做用。

最後，解析器根據 HTTP 方法和頭部來決定是否解析請求體。

路由！

在某種意義上，路由就像是鏈接框架和用戶的橋樑，用戶用合適的方法建立 Router 對象併爲其設置路徑/函數對，而後將它賦值給 App 對象。而 App 對象依次調用 get_handler 函數生成相應的回調函數。簡單來講，路由就負責兩件事，一是存儲路徑/函數對，二是返回須要的路徑/函數對

Router 類中有兩個容許最終開發者添加路由的方法，分別是 add_routes 和 add_route。由於 add_routes 就是 add_route 函數的一層封裝，咱們將主要講解 add_route 函數：

def add_route(self, path, handler):
compiled_route = self.class.build_route_regexp(path)
if compiled_route not in self.routes:
self.routes[compiled_route] = handler
else:
raise DuplicateRoute

首先，這個函數使用 Router.build_router_regexp 的類方法，將一條路由規則（如 '/cars/{id}' 這樣的字符串），「編譯」到一個已編譯的正則表達式對象。這些已編譯的正則表達式用來匹配請求路徑，以及解析開發者自定義的 URL 參數。若是已經存在一個相同的路由，程序就會拋出一個異常。最後，這個路由/處理程序對被添加到一個簡單的字典self.routes中。

下面展現 Router 是如何「編譯」路由的：

@classmethod
def build_route_regexp(cls, regexp_str):
"""
Turns a string into a compiled regular expression. Parses '{}' into
named groups ie. '/path/{variable}' is turned into
'/path/(?P<variable>[a-zA-Z0-9_-]+)'.
:param regexp_str: a string representing a URL path.
:return: a compiled regular expression.
"""
def named_groups(matchobj):
return '(?P<{0}>[a-zA-Z0-9_-]+)'.format(matchobj.group(1))
re_str = re.sub(r'{([a-zA-Z0-9_-]+)}', named_groups, regexp_str)
re_str = ''.join(('^', re_str, '$',))
return re.compile(re_str)

這個方法使用正則表達式將全部出現的 {variable} 替換爲 (?P<variable>)。而後在字符串頭尾分別添加 ^ 和 $ 標記，最後編譯正則表達式對象。

完成了路由存儲僅成功了一半，下面是如何獲得路由對應的函數：

def get_handler(self, path):
logger.debug('Getting handler for: {0}'.format(path))
for route, handler in self.routes.items():
path_params = self.class.match_path(route, path)
if path_params is not None:
logger.debug('Got handler for: {0}'.format(path))
wrapped_handler = HandlerWrapper(handler, path_params)
return wrapped_handler
raise NotFoundException()

一旦 App 對象得到一個 Request 對象，也就得到了 URL 的路徑部分（如 /users/15/edit）。而後，咱們須要匹配函數來生成一個響應或者 404 錯誤。get_handler 函數將路徑做爲參數，循環遍歷路由，對每條路由調用 Router.match_path 類方法檢查是否有已編譯的正則對象與這個請求路徑匹配。若是存在，咱們就調用 HandleWrapper 來包裝路由對應的函數。path_params 字典包含了路徑變量（如 '/users/15/edit' 中的 '15'），若路由沒有指定變量，字典就爲空。最後，咱們將包裝好的函數返回給 App 對象。

若是遍歷了全部的路由都找不到與路徑匹配的，函數就會拋出 NotFoundException 異常。

這個 Route.match 類方法挺簡單：

def match_path(cls, route, path):
match = route.match(path)
try:
return match.groupdict()
except AttributeError:
return None

它使用正則對象的 match 方法來檢查路由是否與路徑匹配。若果不匹配，則返回 None 。

最後，咱們有 HandleWraapper 類。它的惟一任務就是封裝一個異步函數，存儲 path_params 字典，並經過 handle 方法對外提供一個統一的接口。

class HandlerWrapper(object):
def init(self, handler, path_params):
self.handler = handler
self.path_params = path_params
self.request = None
async def handle(self, request):
return await self.handler(request, **self.path_params)

組合到一塊兒

框架的最後部分就是用 App 類把全部的部分聯繫起來。

App 類用於集中全部的配置細節。一個 App 對象經過其 start_server 方法，使用一些配置數據建立一個 HTTPServer 的實例，而後將它傳遞給 asyncio.start_server 函數。asyncio.start_server 函數會對每個 TCP 鏈接調用 HTTPServer 對象的 handle_connection 方法。

def start_server(self):
if not self._server:
self.loop = asyncio.get_event_loop()
self._server = HTTPServer(self.router, self.http_parser, self.loop)
self._connection_handler = asyncio.start_server(
self._server.handle_connection,
host=self.host,
port=self.port,
reuse_address=True,
reuse_port=True,
loop=self.loop)
logger.info('Starting server on {0}:{1}'.format(
self.host, self.port))
self.loop.run_until_complete(self._connection_handler)
try:
self.loop.run_forever()
except KeyboardInterrupt:
logger.info('Got signal, killing server')
except DiyFrameworkException as e:
logger.error('Critical framework failure:')
logger.error(e.traceback)
finally:
self.loop.close()
else:
logger.info('Server already started - {0}'.format(self))

總結

若是你查看源碼，就會發現全部的代碼僅 320 餘行（包括測試代碼的話共 540 餘行）。這麼少的代碼實現了這麼多的功能，讓我有點驚訝。這個框架沒有提供模板、身份認證以及數據庫訪問等功能（這些內容也頗有趣哦）。這也讓我知道，像 Django 和 Tornado 這樣的框架是如何工做的，並且我可以快速地調試它們了。

這也是我按照測試驅動開發完成的第一個項目，整個過程有趣而有意義。先編寫測試用例迫使我思考設計和架構，而不只僅是把代碼放到一塊兒，讓它們能夠運行。不要誤解個人意思，有不少時候，後者的方式更好。不過若是你想給確保這些不怎麼維護的代碼在以後的幾周甚至幾個月依然工做，那麼測試驅動開發正是你須要的。

我研究了下整潔架構以及依賴注入模式，這些充分體如今 Router 類是如何做爲一個更高層次的抽象的（實體？）。Router 類是比較接近核心的，像 http_parser 和 App 的內容比較邊緣化，由於它們只是完成了極小的字符串和字節流、或是中層 IO 的工做。測試驅動開發（TDD）迫使我獨立思考每一個小部分，這使我問本身這樣的問題：方法調用的組合是否易於理解？類名是否準確地反映了我正在解決的問題？個人代碼中是否很容易區分出不一樣的抽象層？

來吧，寫個小框架，真的頗有趣：）

via: http://mattscodecave.com/posts/simple-python-framework-from-scratch.html

做者：Matt 譯者：Cathon 校對：wxy

本文由 LCTT 原創編譯，Linux中國榮譽推出

（題圖來自：es-static.us）