Python Web Flask源碼解讀(一)

關於我
編程界的一名小小程序猿，目前在一個創業團隊任team lead，技術棧涉及Android、Python、Java和Go，這個也是咱們團隊的主要技術棧。 聯繫：hylinux1024@gmail.com

0x00 什麼是WSGI

Web Server Gateway Interface
它由Python標準定義的一套Web Server與Web Application的接口交互規範。

WSGI不是一個應用、框架、模塊或者庫，而是規範。

那什麼是Web Server（Web服務器）和什麼是Web Application（Web 應用）呢？
舉例子來講明容易理解，例如常見的Web應用框架有Django、Flask等，而Web服務器有uWSGI、Gunicorn等。WSGI就是定義了這兩端接口交互的規範。

0x01 什麼是Werkzeug

Werkzeug is a comprehensive WSGI web application library.

Werkzeug是一套實現WSGI規範的函數庫。咱們可使用它來建立一個Web Application（Web應用）。例如本文介紹的Flask應用框架就是基於Werkzeug來開發的。

這裏咱們使用Werkzeug啓動一個簡單的服務器應用

from werkzeug.wrappers import Request, Response

@Request.application
def application(request):
    return Response('Hello, World!')

if __name__ == '__main__':
    from werkzeug.serving import run_simple

    run_simple('localhost', 4000, application)

複製代碼

運行以後能夠在控制檯上將看到以下信息

* Running on http://localhost:4000/ (Press CTRL+C to quit)
複製代碼

使用瀏覽器打開 http://localhost:4000/ 看到如下信息，說明

Hello, World!
複製代碼

0x02 什麼是Flask

Flask is a lightweight WSGI web application framework.

Flask是一個輕量級的web應用框架，它是跑在web服務器中的一個應用。Flask底層就是封裝的Werkzeug。

使用Flask開發一個web應用很是簡單

from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
    return f'Hello, World!'

if __name__ == '__main__':
    app.run()
複製代碼

很簡單吧。

接下來咱們看看Flask應用的啓動流程。

0x03 啓動流程

從項目地址 github.com/pallets/fla… 中把源碼clone下來，而後切換到0.1版本的tag。爲什麼要使用0.1版本呢？由於這個是做者最開始寫的版本，代碼量應該是最少的，並且能夠很容易看到做者總體編碼思路。

下面就從最簡單的Demo開始看看Flask是如何啓動的。咱們知道程序啓動是執行了如下方法

if __name__ == '__main__':
    app.run()
複製代碼

而

app = Flask(__name__)
複製代碼

打開Flask源碼中的__init__方法

Flask.init()

def __init__(self, package_name):
        #: 是否打開debug模式
        self.debug = False

        #: 包名或模塊名
        self.package_name = package_name

        #: 獲取app所在目錄
        self.root_path = _get_package_path(self.package_name)

        #: 存儲視圖函數的字典，鍵爲函數名稱，值爲函數對象，使用@route裝飾器進行註冊
        self.view_functions = {}

        #: 存儲錯誤處理的字典. 鍵爲error code, 值爲處理錯誤的函數，使用errorhandler裝飾器進行註冊
        self.error_handlers = {}

        #: 處理請求前執行的函數列表，使用before_request裝飾器進行註冊
        self.before_request_funcs = []

        #: 處理請求前執行的函數列表，使用after_request裝飾器進行註冊
        self.after_request_funcs = []

        #: 模版上下文
        self.template_context_processors = [_default_template_ctx_processor]
        #: url 映射
        self.url_map = Map()
        
        #: 靜態文件
        if self.static_path is not None:
            self.url_map.add(Rule(self.static_path + '/<filename>',
                                  build_only=True, endpoint='static'))
            if pkg_resources is not None:
                target = (self.package_name, 'static')
            else:
                target = os.path.join(self.root_path, 'static')
            self.wsgi_app = SharedDataMiddleware(self.wsgi_app, {
                self.static_path: target
            })

        #: 初始化 Jinja2 模版環境. 
        self.jinja_env = Environment(loader=self.create_jinja_loader(),
                                     **self.jinja_options)
        self.jinja_env.globals.update(
            url_for=url_for,
            get_flashed_messages=get_flashed_messages
        )
複製代碼

在Flask的構造函數中進行了各類初始化操做。

而後就是執行app.run()方法

app.run()

def run(self, host='localhost', port=5000, **options):
    """啓動本地開發服務器. 若是debug設置爲True，那麼會自動檢查代碼是否改動，有改動則會自動執行部署 :param host: 監聽的IP地址. 若是設置爲 ``'0.0.0.0'``就能夠進行外部訪問 :param port: 端口，默認5000 :param options: 這個參數主要是對應run_simple中須要的參數 """
    from werkzeug.serving import run_simple
    if 'debug' in options:
        self.debug = options.pop('debug')
    options.setdefault('use_reloader', self.debug)
    options.setdefault('use_debugger', self.debug)
    return run_simple(host, port, self, **options)
複製代碼

run很簡潔，主要是調用了werkzeug.serving中的run_simple方法。

再打開run_simple的源碼

rum_simple()

def run_simple(hostname, port, application, use_reloader=False, use_debugger=False, use_evalex=True, extra_files=None, reloader_interval=1, threaded=False, processes=1, request_handler=None, static_files=None, passthrough_errors=False, ssl_context=None):
    # 這方法仍是比較短的，可是註釋寫得很詳細，因爲篇幅問題，就把源碼中的註釋省略了
    if use_debugger:
        from werkzeug.debug import DebuggedApplication
        application = DebuggedApplication(application, use_evalex)
    if static_files:
        from werkzeug.wsgi import SharedDataMiddleware
        application = SharedDataMiddleware(application, static_files)

    def inner():
        make_server(hostname, port, application, threaded,
                    processes, request_handler,
                    passthrough_errors, ssl_context).serve_forever()

    if os.environ.get('WERKZEUG_RUN_MAIN') != 'true':
        display_hostname = hostname != '*' and hostname or 'localhost'
        if ':' in display_hostname:
            display_hostname = '[%s]' % display_hostname
        _log('info', ' * Running on %s://%s:%d/', ssl_context is None
             and 'http' or 'https', display_hostname, port)
    if use_reloader:
        # Create and destroy a socket so that any exceptions are raised before
        # we spawn a separate Python interpreter and lose this ability.
        test_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        test_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        test_socket.bind((hostname, port))
        test_socket.close()
        run_with_reloader(inner, extra_files, reloader_interval)
    else:
        inner()
複製代碼

在rum_simple方法中還定義一個嵌套方法inner()，這個是方法的核心部分。

inner()

def inner():
    make_server(hostname, port, application, threaded, processes, request_handler, passthrough_errors, ssl_context).serve_forever()
複製代碼

在inner()方法裏面，調用make_server(...).serve_forever()啓動了服務。

make_server()

def make_server(host, port, app=None, threaded=False, processes=1, request_handler=None, passthrough_errors=False, ssl_context=None):
    """Create a new server instance that is either threaded, or forks or just processes one request after another. """
    if threaded and processes > 1:
        raise ValueError("cannot have a multithreaded and "
                         "multi process server.")
    elif threaded:
        return ThreadedWSGIServer(host, port, app, request_handler,
                                  passthrough_errors, ssl_context)
    elif processes > 1:
        return ForkingWSGIServer(host, port, app, processes, request_handler,
                                 passthrough_errors, ssl_context)
    else:
        return BaseWSGIServer(host, port, app, request_handler,
                              passthrough_errors, ssl_context)
複製代碼

在make_server()中會根據線程或者進程的數量建立對應的WSGI服務器。Flask在默認狀況下是建立BaseWSGIServer服務器。

BaseWSGIServer、ThreadedWSGIServer、ForkingWSGIServer

class BaseWSGIServer(HTTPServer, object):
    ...

class ThreadedWSGIServer(ThreadingMixIn, BaseWSGIServer):
    """A WSGI server that does threading."""
    ...

class ForkingWSGIServer(ForkingMixIn, BaseWSGIServer):
    """A WSGI server that does forking."""
    ...
複製代碼

能夠看出他們以前的繼承關係以下

打開BaseWSGIServer的start_server()方法

start_server()

def serve_forever(self):
    try:
        HTTPServer.serve_forever(self)
    except KeyboardInterrupt:
        pass
複製代碼

能夠看到最終是使用HTTPServer中的啓動服務的方法。而HTTPServer是Python標準類庫中的接口。

HTTPServer是socketserver.TCPServer的子類

socketserver.TCPServer

若是要使用Python中類庫啓動一個http server，則相似代碼應該是這樣的

import http.server
import socketserver

PORT = 8000

Handler = http.server.SimpleHTTPRequestHandler

with socketserver.TCPServer(("", PORT), Handler) as httpd:
    print("serving at port", PORT)
    httpd.serve_forever()
複製代碼

至此，整個服務的啓動就到這裏就啓動起來了。

這個過程的調用流程爲

graph TD

A[Flask]-->B[app.run]
B[app.run]-->C[werkzeug.run_simple]
C[werkzeug.run_simple]-->D[BaseWSGIServer]
D[BaseWSGIServer]-->E[HTTPServer.serve_forever]
E[HTTPServer.serve_forever]-->F[TCPServer.serve_forever]
複製代碼

0x04 總結一下

WSGI是WEB服務器與WEB應用之間交互的接口規範。werkzeug是實現了這一個規範的函數庫，而Flask框架是基於werkzeug來實現的。 咱們從Flask.run()方法啓動服務開始，追蹤了整個服務啓動的流程。

0x05 學習資料

• werkzeug.palletsprojects.com/en/0.15.x/
• palletsprojects.com/p/flask/
• docs.python.org/3/library/h…