应用程序调度¶
应用程序调度是在wsgi级别上组合多个Flask应用程序的过程。您不仅可以组合flask应用程序,还可以组合任何wsgi应用程序。这将允许您在同一个解释器中并排运行django和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应用程序,该应用程序查看到来的请求并将其委托给您的FlaskTM应用程序。如果该应用程序尚不存在,则会动态创建并记住它。
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)