应用程序分发

应用程序分发是指在 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 应用,Dispatcher Middleware 的作用是根据 URL 前缀将它们组合成一个更大的应用程序进行分发。

例如,你可以让主应用运行在 /,而后台接口运行在 /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)