跨flask app使用url_for

用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))

Post Views: 834