8/21/2018 - 6:13 AM

werkzeug.wsgi.DispatcherMiddleware() 的几种应用实例

[Python Web] Pyhon 后端中的一些代码记录 #Python #Backend

# 1、

def run(self):

        app_root = self.flask.config.get("APPLICATION_ROOT")
        if app_root:
            def simple(env, resp):
                resp(b'200 OK', [(b'Content-Type', b'text/html')])
                return [b'<a href="{root}">{root}</a>'.format(root=app_root)]
            print self.flask.config["APPLICATION_ROOT"]
            self.flask.wsgi_app = DispatcherMiddleware(simple, {
                self.flask.config["APPLICATION_ROOT"]: self.flask.wsgi_app
        # Add extra files
        extra_files = []
        for (root, _, files) in os.walk(os.path.join(const.PUSH_WEB_DIR, "templates")):
            extra_files.extend(os.path.join(root, f) for f in files), self.port, debug=self.debug, extra_files=extra_files)

# 2、

def __init__(self, url, service_impl_cls):
        self.url = url_endswith_slash(url)
        self.scheme,, self.path, _, _ = urllib.parse.urlsplit(

        def null_app(environ, start_response):
            start_response('404 Not Found', [('Content-Type', 'text/plain')])
            return ['Not Found']

        wsgi_app = WsgiApp(service_impl_cls())
        if self.path != '/':
            self.wsgi_app = DispatcherMiddleware(
                null_app, {self.path[:-1]: wsgi_app}
            self.wsgi_app = wsgi_app
        self.wsgi_test_client = Client(self.wsgi_app, Response)

# 3、

def start_app():
    app.config['DEBUG'] = args.debug
    app.config['args'] = args
    if args.root and args.root != '/':
        if not args.root.startswith('/'):
            args.root = '/' + args.root
        from werkzeug.serving import run_simple
        from werkzeug.wsgi import DispatcherMiddleware
        app.config["APPLICATION_ROOT"] = args.root
        application = DispatcherMiddleware(simple, {
            app.config['APPLICATION_ROOT']: app,
        run_simple(args.listen, args.port, application, use_reloader=args.debug)
    else:, args.port, debug=args.debug)

# 4、

def main():
  ''' Runs the Flux Flask application and all required components. '''

    # Test if Git version is at least 2.3 (for GIT_SSH_COMMAND)
    git_version = subprocess.check_output(['git', '--version']).decode().strip()
    git_version ='^git version (\d\.\d+)', git_version)
    if git_version:
      git_version =
    if not git_version or int(git_version.split('.')[1]) < 3:
      print('Error: {!r} installed but need at least 2.3'.format(git_version))

    # Make sure the root user exists and has all privileges, and that
    # the password is up to date.
    with models.Session() as session:

    # Create a dispatcher for the sub-url under which the app is run.
    url_prefix = urlparse(config.app_url).path
    if url_prefix and url_prefix != '/':
      from werkzeug.wsgi import DispatcherMiddleware
      target_app = DispatcherMiddleware(flask.Flask('_dummy_app'), {
        url_prefix: app,
      target_app = app

    print(' * starting builder threads...')
      from werkzeug.serving import run_simple
      run_simple(, config.port, target_app, use_reloader=False)
      print(' * stopping builder threads...')

# 5、

def create_app(cfg_files=['DEFAULT']):
    """Factory for making the web Flask application

    :param cfg_files: Single or more config file(s)
    :return: Constructed web application
    :rtype: ``repocribro.repocribro.Repocribro``
    app = Repocribro()
    from .database import db
    ext_master = ExtensionsMaster(app=app, db=db)
    app.container.set_singleton('ext_master', ext_master)

    if cfg_files == ['DEFAULT']:
        cfg_files = os.environ.get('REPOCRIBRO_CONFIG_FILE',

    config = create_config(cfg_files)
    config.set('flask', 'release', RELEASE)
    app.container.set_singleton('config', config)'setup_config')

    app.secret_key = config.get('flask', 'secret_key')

    app.container.set_singleton('db', db)

    ext_names ='introduce', 'unknown')
    print('Loaded extensions: {}'.format(', '.join(ext_names)))'init_first')'init_models')'init_business')'init_filters')'init_blueprints')'init_container')

    if config.has_option('flask', 'application_root'):
        from werkzeug.serving import run_simple
        from werkzeug.wsgi import DispatcherMiddleware
        app.wsgi_app = DispatcherMiddleware(
            {config.get('flask', 'application_root'): app.wsgi_app}

    return app

# 6、Flask

def run(port=5000, mongo_host='localhost', mongo_port=27017, pagination_limit=1000000):
    client = j.clients.osis.getByInstance('main')

    apps = dict()
    for namespace in client.listNamespaces():
        dbname = namespace if namespace != 'system' else 'js_system'
        my_settings = {
                'MONGO_HOST': mongo_host,
                'MONGO_PORT': mongo_port ,
                'MONGO_DBNAME': dbname,
                'DOMAIN': generateDomain(spec),
                'RESOURCE_METHODS': ['GET', 'POST'],
                'ITEM_METHODS': ['GET', 'PATCH', 'PUT', 'DELETE'],
                'X_DOMAINS': '*',
                'MONGO_QUERY_BLACKLIST': [],
                'X_HEADERS': ["X-HTTP-Method-Override", 'If-Match'],
                'PAGINATION_LIMIT': pagination_limit
        # init application
        app = Eve(__name__, settings=my_settings)
        def ui():
            return render_template('ui.html')
        # Unfortunately, eve_docs doesn't support CORS (too bad!), so we had to reimplement it ourselves
        def specs():
            return send_response(None, [get_cfg()])
        apps['/%s' % namespace] = app
    print "visit:\nhttp://localhost:%s/docs/" % port
    if apps:
        firstapp = apps.values()[0]
        application = DispatcherMiddleware(firstapp, apps)
    # let's roll
        run_simple('', port, application, use_reloader=False)

# 7、Tornado

def run(self, host=None, port=None, debug=None, **options):
        import tornado.wsgi
        import tornado.ioloop
        import tornado.httpserver
        import tornado.web

        if host is None:
            host = ''
        if port is None:
            server_name = self.config['SERVER_NAME']
            if server_name and ':' in server_name:
                port = int(server_name.rsplit(':', 1)[1])
                port = 5000
        if debug is not None:
            self.debug = bool(debug)

        hostname = host
        port = port
        application = self
        use_reloader = self.debug
        use_debugger = self.debug

        if use_debugger:
            from werkzeug.debug import DebuggedApplication
            application = DebuggedApplication(application, True)

            from .webdav import dav_app
        except ImportError as e:
            logger.warning('WebDav interface not enabled: %r', e)
            dav_app = None
        if dav_app:
            from werkzeug.wsgi import DispatcherMiddleware
            application = DispatcherMiddleware(application, {
                '/dav': dav_app

        container = tornado.wsgi.WSGIContainer(application)
        self.http_server = tornado.httpserver.HTTPServer(container)
        self.http_server.listen(port, hostname)
        if use_reloader:
            from tornado import autoreload
            autoreload.start()'webui running on %s:%s', hostname, port)
        self.ioloop = tornado.ioloop.IOLoop.current()