应用分发

应用分发是在 WSGI 层面组合多个 Flask 应用的过程。你不仅可以组合 Flask 应用,还可以组合任何 WSGI 应用。如果你愿意,这允许你在同一个解释器中并行运行 Django 和 Flask 应用。这的实用性取决于应用内部的工作方式。

大型应用作为包的根本区别在于,在这种情况下,你运行的是相同或不同的 Flask 应用,它们彼此完全隔离。它们运行不同的配置,并在 WSGI 层面进行分发。

使用本文档

以下每种技术和示例都会生成一个 application 对象,该对象可以使用任何 WSGI 服务器运行。对于开发,使用 flask run 命令来启动开发服务器。对于生产,请参阅部署到生产

from flask import Flask

app = Flask(__name__)

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

组合应用

如果你有完全分离的应用,并且希望它们在同一个 Python 解释器进程中彼此相邻工作,你可以利用 werkzeug.wsgi.DispatcherMiddleware。这里的想法是,每个 Flask 应用都是一个有效的 WSGI 应用,它们通过分发器中间件组合成一个更大的应用,该应用基于前缀进行分发。

例如,你可以让你的主应用在 / 上运行,而你的后端接口在 /backend 上运行。

from werkzeug.middleware.dispatcher import DispatcherMiddleware
from frontend_app import application as frontend
from backend_app import application as backend

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

按子域名分发

有时你可能想要使用同一应用的多个实例,但配置不同。假设应用是在函数内部创建的,你可以调用该函数来实例化它,这非常容易实现。为了开发你的应用以支持在函数中创建新实例,请查看应用工厂模式。

一个非常常见的例子是为每个子域名创建应用。例如,你配置你的 Web 服务器将所有子域名的所有请求分发到你的应用,然后你使用子域名信息来创建用户特定的实例。一旦你设置好服务器以监听所有子域名,你就可以使用一个非常简单的 WSGI 应用来完成动态应用创建。

在这方面,抽象的完美级别是 WSGI 层。你编写自己的 WSGI 应用,它查看传入的请求并将其委托给你的 Flask 应用。如果该应用尚不存在,则会动态创建并记住它。

from threading import Lock

class SubdomainDispatcher:

    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)

然后可以像这样使用此分发器

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)

按路径分发

按 URL 上的路径分发非常相似。与其查看 Host 标头来确定子域名,不如简单地查看请求路径,直到第一个斜杠。

from threading import Lock
from wsgiref.util import shift_path_info

class PathDispatcher:

    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:
            shift_path_info(environ)
        else:
            app = self.default_app
        return app(environ, start_response)

def _peek_path_info(environ):
    segments = environ.get("PATH_INFO", "").lstrip("/").split("/", 1)
    if segments:
        return segments[0]

    return None

这与子域名分发的主要区别在于,如果创建器函数返回 None,则此分发会回退到另一个应用。

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)