一.什么是WSGI

　　WSGI全稱Web Server Gateway Interface, 即web服務器網關介面.

　　如圖所示的http服務器, web服務器負責面向客戶端接受請求和回傳回應, web應用負責處理請求, 因此二者需要進行互動, web服務器將請求資料傳給web應用, web應用執行完業務邏輯后將處理結果回傳給web服務器. WSGI就是通過標準介面的方式, 定義了web服務器和python web應用之間的互動規范. 對于一個python web應用, 只要它遵循WSGI規范, 理論上就可以在任何寫入這個規范的服務器上運行, 這樣就提高了程式的可移植性.

　　WSGI的介面定義很簡單, 對于web應用端, 只需要定義一個類似如下的可呼叫物件:

from typing import List, Callable


def my_application(environ: dict, start_response: Callable) -> List[bytes]:
    # do something
    status = '200 OK'
    headers = [('Content-Type', 'text/html')]
    start_response(status, headers)
    return [b'<h1>hello wsgi</h1>']

　　這個可呼叫物件要接收兩個引數. 第一個引數environ是一個字典, 內容是客戶端的請求資訊, 像是下面這樣:

　　第二個引數start_response則是一個函式, 它接收status和headers兩個引數, 當后端業務邏輯執行完畢之后, 呼叫它就可以設定回應狀態和回應頭.

　　最后, web應用需要將回應體以位元組的形式放在一個可迭代物件中回傳.

　　這樣, WSGI就定義了一個web應用的行為模式: 首先決議environ提取需要的資料, 進行各種操作后, 呼叫start_response函式設定回應狀態和訊息報頭, 最后將回應正文放在可迭代物件中回傳.
　　對于web服務器端,使用如下方式就能啟動服務并呼叫剛才寫的web應用:

from wsgiref.simple_server import make_server
from app import my_application #這里的app表示web應用所在的地方, 根據自己的代碼自行修改

httpd = make_server('127.0.0.1', 8000, my_application)
httpd.serve_forever()

　　這樣, 一個最簡單的http服務端就寫好了. 訪問127.0.0.1:8000, 得到結果如下:

二.web框架的架構

　　借用Django中MTV的那一套, 一個web框架主要包含路由, 控制器View, 模板層Template和模型層Model. web服務器將資料傳給路由, 路由再將資料分發給對應的控制器去處理. 控制器與模型層互動得到資料庫資料, 與模板層互動得到前端網頁檔案后, 將處理結果回傳給web服務器.

　　基于上面的一套流程, 本文要實作的web框架需要實作下面的一些功能:

Request和Response類:

　　分別作為environ引數和回應引數的封裝, 這樣可以讓代碼更加清晰和整潔.

Route類:

　　負責路由的分發.

Application類:

　　一方面作為WSGI的呼叫入口, 一方面進行整體的調度.

Template:

　　渲染html文本, 生成動態頁面

Model:

　　這個不寫了,因為我對資料庫的了解還不夠.

三.Request和Response類

　　Request類的代碼如下:

from urllib.parse import parse_qs
from typing import List


class Request:

    def __init__(self, environ: dict) -> None:
        self.environ = environ
        self.method = environ['REQUEST_METHOD']
        self.path = environ['PATH_INFO']
        # 首先獲取url中攜帶的引數
        self.arguments = parse_qs(environ['QUERY_STRING'])
        # 然后讀取form表單引數
        try:
            request_body_size = int(environ['CONTENT_LENGTH'])
        except (ValueError, KeyError):
            request_body_size = 0
        request_body = environ['wsgi.input'].read(request_body_size).decode()
        # 最后將二者整合,表單資料中相同key的引數會把url的覆寫
        self.arguments.update(parse_qs(request_body))

    def __str__(self) -> str:
        items = (
            'arguments',
            'path',
            'method',
        )
        info = {item: getattr(self, item) for item in items}
        return str(info)

    def get_argument(self, key: str, default: str = '') -> str:
        values = self.arguments.get(key, None)
        if not values:
            return default
        return values[0]

    def get_arguments(self, key: str) -> List[str]:
        values = self.arguments.get(key, None)
        if values is None:
            return []
        return values

　　Request類作為environ的封裝. 這里只提取最常用的幾個資訊: 請求型別, 請求路徑和url以及form表單的引數, 其他資訊本文用不上, 就不提取了.

　　Response類的代碼如下:

import http.client
from wsgiref.headers import Headers
from typing import Iterable, List, Tuple


class Response:

    def __init__(self, body: Iterable[str] = None, status_code: int = 200, content_type: str = 'text/html',
                 charset: str = 'utf-8') -> None:
        if body is None:
            self.body = []
        elif isinstance(body, str):
            self.body = [body]
        else:
            self.body = body
        self.status_code = status_code
        self._headers = Headers()
        self._headers.add_header('content-type', f'{content_type}; charset={charset}')
        self.charset = charset

    def __iter__(self) -> Iterable[bytes]:
        for val in self.body:yield val.encode(self.charset)

    @property
    def status(self) -> str:
        status_string = http.client.responses.get(self.status_code, 'UNKNOWN')
        return f'{self.status_code} {status_string}'

    @property
    def headers(self) -> List[Tuple[str, str]]:
        return self._headers.items()

　　Response類和Request類類似, 封裝一些回應的基本的資訊. 由于WSGI定義的介面要求回傳一個可迭代物件, 因此這里定義response的__iter__方法, 將回應資料轉碼后回傳.

　　通過如下方式使用上面定義的兩個類:

from typing import Callable, Iterable


def application(environ: dict, start_response: Callable) -> Iterable[bytes]:
    request = Request(environ)
    print(request)
    response = Response(['<h1>hello wsgi</h1>'])
    start_response(response.status, response.headers)
    return response

四.路由分發

　　路由的寫法參照了flask, 通過裝飾器的形式來指定路由分發. 代碼如下:

import re
from typing import (
    Callable,
    Sequence,
    Tuple,
)


class Route:

    def __init__(self) -> None:
        self.handlers = []
        for pattern in (r'/(favicon\.ico)', r'/(robots\.txt)',):
            self.handlers.append((re.compile(pattern), static_handler))

    def __call__(self, pattern: str) -> Callable:
        if not pattern.endswith('$'):
            pattern += '$'

        def wrapper(handler: Callable) -> None:
            self.handlers.append((re.compile(pattern), handler))

        return wrapper

    def match(self, path: str) -> Tuple[Callable, Sequence]:
        for pattern, handler in self.handlers:
            args = re.match(pattern, path)
            if args:
                return handler, args.groups()
        raise HttpError(404)

　　路由需要保存不同的url和對應的handler, 因此這里使用類而不是函式來作為裝飾器, 使用handlers這個屬性來保存url和對應的handler函式. 當接收到客戶端的請求時, 呼叫路由物件的match方法就能得到對應的處理函式和url中的匹配資料. 如果無法匹配, 則通過HttpError類來引發404錯誤. HttpError類的代碼如下:

import http.client


class HttpError(Exception):
    def __init__(self, status_code: int = 500, message: str = '') -> None:
        self.status_code = status_code
        self.message = message

    def __str__(self) -> str:
        error = f'HTTP {self.status_code}: {http.client.responses.get(self.status_code, "Unknown")}'
        if self.message:
            return error + f'({self.message})'
        else:
            return error

　　HttpError的本質就是一個包含了狀態碼和錯誤資訊的Exception, 實作這個類的好處是, 當某個流程無法繼續的時候, 可以不用傳引數, 而是直接將這個例外拋給上層.

　　另外, Route類在初始化的時候, 在路由串列中加了r'/(favicon\.ico)'和r'/(robots\.txt)'兩項, 這兩分別是網站的圖示和爬蟲協議, 因此交由static_handler函式處理:

def static_handler(request: Request, file_name: str) -> None:
    # 本文用不上,直接404得了
    raise HttpError(404)

　　最后, 通過下面的代碼測驗剛才寫的路由:

route = Route()


@route(r'/')
def index(request: Request) -> str:
    return 'hello wsgi'


@route(r'/(.*)')
def hello(request: Request, username: str) -> str:
    return f'hello {username}'


def application(environ: dict, start_response: Callable) -> Iterable[bytes]:
    request = Request(environ)
    try:
        handler, args = route.match(request.path)
        response = Response(handler(request, *args))
    except HttpError as e:
        response = Response(status_code=e.status_code)
    start_response(response.status, response.headers)
    return response

五.Application

　　application的本質就是WSGI的web應用端入口. 為了方便擴展, 這里把它寫成類的形式, 然后定義__call__方法;

from wsgiref.simple_server import make_server
from typing import Callable, Iterable


class Application:

    def __init__(self) -> None:
        self.route = Route()

    def __call__(self, environ: dict, start_response: Callable) -> Iterable[bytes]:
        request = Request(environ)
        try:
            handler, args = self.route.match(request.path)
            response = Response(handler(request, *args))
        except HttpError as e:
            response = Response(status_code=e.status_code)
        start_response(response.status, response.headers)
        return response

    def run(self, host: str = '127.0.0.1', port: int = 8000) -> None:
        httpd = make_server(host, port, self)
        httpd.serve_forever()

　　沒什么好說的, 和之前的代碼是一樣的.

　　這樣一來, 一個web框架就差不多成型了, 用下面的代碼進行測驗:

app = Application()


@app.route(r'/')
def index(request: Request) -> str:
    return 'hello wsgi'


@app.route(r'/(.*)')
def hello(request: Request, username: str) -> str:
    return f'hello {username}'


if __name__ == '__main__':
    app.run()

六.模板層

　　關于如何實作模板層, 我單獨寫了一篇博客, 在這里.

七.模型層

　　能力不夠, 先不講了.

八.結束

　　把上面的代碼整合起來, 我們就得到了一個web框架, 用起來像是這樣:

from framework import Application, Request, render

app = Application()


@app.route(r'/(.*)')
def hello(request: Request, username: str) -> str:
    return render('<h1>hello {{username}}</h1>', {'username': username})

　　呼叫app.run, 就可以運行hello應用了. 此外, 也可以使用其他支持WSGI的web服務器來啟動服務, 比如gunicorn:

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/141154.html

標籤：Python

上一篇：基于Python豆瓣自動化測驗【2】

下一篇：學無止下載器破解

寫個基于WSGI的web框架