Flask 應(yīng)用調(diào)度

2021-08-11 09:47 更新

應(yīng)用調(diào)度指的是在 WSGI 層次合并運(yùn)行多個 Flask 的應(yīng)用的進(jìn)程。您不能將 Flask 與更大的東西合并,但是可以和 WSGI 應(yīng)用交叉。這甚至允許您將 Django 和 Flask 的應(yīng)用運(yùn)行在同一個解釋器下。這么做的用處依賴于 這個應(yīng)用內(nèi)部是如何運(yùn)行的。

模塊方式 的區(qū)別在于,此時您運(yùn)行的不 同 Flask 應(yīng)用是相互之間完全獨(dú)立的,他們運(yùn)行在不同的配置,而且在 WSGI 層調(diào)度。

如何使用此文檔

下面的所有技巧和例子都將最終得到一個 application 對象,這個對象 可以在任何 WSGI 服務(wù)器上運(yùn)行。在生產(chǎn)環(huán)境下,請參看 部署選擇 相關(guān)章節(jié)。在開發(fā)時,Werkzeug 提供了一個提供了一個內(nèi)置的開發(fā)服務(wù)器, 可以通過 werkzeug.serving.run_simple() 函數(shù)使用:

from werkzeug.serving import run_simple
run_simple('localhost', 5000, application, use_reloader=True)

注意,run_simple 函數(shù)不是為生產(chǎn) 用途設(shè)計(jì)的,發(fā)布應(yīng)用時可以使用 成熟的 WSGI 服務(wù)器

為了能使用交互式調(diào)試器,調(diào)試必須在應(yīng)用和簡易開發(fā)服務(wù)器兩邊都被激活。 下面是一個帶有調(diào)試功能的 “Hello World” 的例子:

from flask import Flask
from werkzeug.serving import run_simple

app = Flask(__name__)
app.debug = True

@app.route('/')
def hello_world():
    return 'Hello World!'

if __name__ == '__main__':
    run_simple('localhost', 5000, app,
               use_reloader=True, use_debugger=True, use_evalex=True)

合并應(yīng)用

如果您有一些完全獨(dú)立的應(yīng)用程序,而您希望他們使用同一個 Python 解釋器, 背靠背地運(yùn)行,您可以利用 werkzeug.wsgi.DispatcherMiddleware 這個類。 這里,每個 Flask 應(yīng)用對象都是一個有效的 WSGI 應(yīng)用對象,而且他們在 調(diào)度中間層當(dāng)中被合并進(jìn)入一個規(guī)模更大的應(yīng)用,并通過前綴來實(shí)現(xiàn)調(diào)度。

例如,您可以使您的主應(yīng)用運(yùn)行在 / 路徑,而您的后臺 接口運(yùn)行在 /backend 路徑:

from werkzeug.wsgi import DispatcherMiddleware
from frontend_app import application as frontend
from backend_app import application as backend

application = DispatcherMiddleware(frontend, {
    '/backend':     backend
})

通過子域名調(diào)度

有時,您希望使用對一個應(yīng)用使用不同的配置,對每個配置運(yùn)行一個實(shí)例,從而有 多個實(shí)例存在。假設(shè)應(yīng)用對象是在函數(shù)中生成的,您就可以調(diào)用這個函數(shù)并實(shí)例化 一個實(shí)例,這相當(dāng)容易實(shí)現(xiàn)。為了使您的應(yīng)用支持在函數(shù)中創(chuàng)建新的對象,請先參考 應(yīng)用程序的工廠函數(shù) 模式。

一個相當(dāng)通用的例子,那就是為不同的子域名創(chuàng)建不同的應(yīng)用對象。比如 您將您的Web服務(wù)器設(shè)置為將所有的子域名都分發(fā)給您的引用,而您接下來 使用這些子域名信息創(chuàng)建一個針對特定用戶的實(shí)例。一旦您使得您的服務(wù)器 偵聽所有的子域名請求,那么您就可以使用一個非常簡單的 WSGI 對象 來進(jìn)行動態(tài)的應(yīng)用程序構(gòu)造。

實(shí)現(xiàn)此功能最佳的抽象層就是 WSGI 層。您可以編寫您自己的 WSGI 程序來 檢查訪問請求,然后分發(fā)給您的 Flask 應(yīng)用。如果您的應(yīng)用尚未存在,那么 就創(chuàng)建一個并且保存下來:

from threading import Lock

class SubdomainDispatcher(object):

    def __init__(self, domain, create_app):
        self.domain = domain
        self.create_app = create_app
        self.lock = Lock()
        self.instances = {}

    def get_application(self, host):
        host = host.split(':')[0]
        assert host.endswith(self.domain), 'Configuration error'
        subdomain = host[:-len(self.domain)].rstrip('.')
        with self.lock:
            app = self.instances.get(subdomain)
            if app is None:
                app = self.create_app(subdomain)
                self.instances[subdomain] = app
            return app

    def __call__(self, environ, start_response):
        app = self.get_application(environ['HTTP_HOST'])
        return app(environ, start_response)

調(diào)度器可以這樣使用:

from myapplication import create_app, get_user_for_subdomain
from werkzeug.exceptions import NotFound

def make_app(subdomain):
    user = get_user_for_subdomain(subdomain)
    if user is None:
        # if there is no user for that subdomain we still have
        # to return a WSGI application that handles that request.
        # We can then just return the NotFound() exception as
        # application which will render a default 404 page.
        # You might also redirect the user to the main page then
        return NotFound()

    # otherwise create the application for the specific user
    return create_app(user)

application = SubdomainDispatcher('example.com', make_app)

使用路徑來調(diào)度

通過 URL 路徑分發(fā)請求跟前面的方法很相似。只需要簡單檢查請求路徑當(dāng)中到第一個 斜杠之前的部分,而不是檢查用來確定子域名的 HOST 頭信息就可以了:

from threading import Lock
from werkzeug.wsgi import pop_path_info, peek_path_info

class PathDispatcher(object):

    def __init__(self, default_app, create_app):
        self.default_app = default_app
        self.create_app = create_app
        self.lock = Lock()
        self.instances = {}

    def get_application(self, prefix):
        with self.lock:
            app = self.instances.get(prefix)
            if app is None:
                app = self.create_app(prefix)
                if app is not None:
                    self.instances[prefix] = app
            return app

    def __call__(self, environ, start_response):
        app = self.get_application(peek_path_info(environ))
        if app is not None:
            pop_path_info(environ)
        else:
            app = self.default_app
        return app(environ, start_response)

這種例子與之前子域名調(diào)度那里的區(qū)別是,這里如果創(chuàng)建應(yīng)用對象的函數(shù)返回了 None, 那么請求就被降級回推到另一個應(yīng)用當(dāng)中:

from myapplication import create_app, default_app, get_user_for_prefix

def make_app(prefix):
    user = get_user_for_prefix(prefix)
    if user is not None:
        return create_app(user)

application = PathDispatcher(default_app, make_app)


以上內(nèi)容是否對您有幫助:
在線筆記
App下載
App下載

掃描二維碼

下載編程獅App

公眾號
微信公眾號

編程獅公眾號