跨flask app使用url_for

2018-01-26 10:33:07来源:http://jinzhao.me/archives/354作者:小核桃人点击

分享

用flask的肯定知道url_for,解放了记住url的繁琐。前阵子因为flask臭名昭着的 servername问题彻底拆分了应用程序,在解决了大部分引用和重用后发现应用间导航也是个问题,写硬url显然不科学。这里祭出几大法宝用来解决这个问题:


中间件Dispatcher,也可以使用werkzeug.wsgi 的 DispatcherMiddleware,中间件的设计理念真是扩展所必须的呀;
with nebapp.app_context,没错在每个应用程序中创建邻居app后可以使用它的上下文即可用url_for了;
实际使用会有报错,这里需要mock一个request,这里flask准备好了app.test_request_context;

贴一下我在用的Dispatcher


class Dispatcher:
"""
Allows one to mount middlewares or applications in a WSGI application.
This is useful if you want to combine multiple WSGI applications::
app = DispatcherMiddleware(app, {
'/app2':app2,
'/app3':app3
})
"""
def __init__(self, app, mounts=None):
self.app = app
self.mounts = mounts or {}
self.url_for_resolver = URLForResolver(
self.mounts
)
self.app.dispatcher = self
def __call__(self, environ, start_response):
script = environ.get('PATH_INFO', '')
path_info = ''
while '/' in script:
if script in self.mounts:
app = self.mounts[script]
break
script, last_item = script.rsplit('/', 1)
path_info = '/%s%s' % (last_item, path_info)
else:
app = self.mounts.get(script, self.app)
original_script_name = environ.get('SCRIPT_NAME', '')
environ['SCRIPT_NAME'] = original_script_name + script
# Convert empty path info values to a forward slash '/'
environ['PATH_INFO'] = path_info or '/'
return app(environ, start_response)
class URLForResolver:
"""
A URL resolver that provides resolution of `url_for` across multiple apps.
"""
def __init__(self, mounts):
self.mounts=mounts
self.apps = list(mounts.values())
self.cache = {}
for app in self.apps:
app.url_build_error_handlers.append(self)
def __call__(self, endpoint, **values):
"""Attempt to resolve a URL any of the registered apps"""
error = Exception('can not find endpoint for %s' % endpoint)
parts=endpoint.split('.')
path=parts.pop()
if len(parts)>1:
path='%s.%s'%(parts[1],path)
app_name=None
if len(parts)>0:
app_name=parts[0]
# Check if we have a cached look up
if endpoint in self.cache:
app = self.cache[endpoint]
if app:
with app.app_context(), app.test_request_context():
ret_path= url_for(path, **values)
if ':5000' in ret_path:
ret_path=ret_path.replace(':5000','')
return ret_path
else:
raise error
if app_name:
app = self.mounts[app_name]
self.cache[endpoint] = app
return self(endpoint, **values)
# Attempt to find an app with the registered endpoint
for app in self.apps:
for rule in app.url_map.iter_rules():
if rule.endpoint == path:
# Found - cache the result and call self to return the URL
self.cache[endpoint] = app
return self(endpoint, **values)
# Not found - cache the result and re-raise the error
self.cache[endpoint] = None
raise error

使用方式:


#初始化
init_apps()
apps = create_apps()
del apps['/passport']
dispatcher = Dispatcher(passport, mounts=apps)
#调用
current_app.dispatcher.url_for_resolver('/appname.blueprint_name.page_name',_external=True))

微信扫一扫

第七城市微信公众平台