python werkzeug.wsgi.DispatcherMiddleware（）的几种应用实例

Posted 2021-05-09

# 1、

def run(self):
        views.create_views(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.flask.run(self.host, 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.host, self.path, _, _ = urllib.parse.urlsplit(
            self.url
        )

        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}
            )
        else:
            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:
        app.run(args.listen, 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 = re.search('^git version (\d\.\d+)', git_version)
    if git_version:
      git_version = git_version.group(1)
    if not git_version or int(git_version.split('.')[1]) < 3:
      print('Error: {!r} installed but need at least 2.3'.format(git_version))
      sys.exit(1)

    # Make sure the root user exists and has all privileges, and that
    # the password is up to date.
    with models.Session() as session:
      models.User.create_or_update_root(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 != '/':
      print(url_prefix)
      from werkzeug.wsgi import DispatcherMiddleware
      target_app = DispatcherMiddleware(flask.Flask('_dummy_app'), {
        url_prefix: app,
      })
    else:
      target_app = app

    print(' * starting builder threads...')
    build.run_consumers(num_threads=config.parallel_builds)
    build.update_queue()
    try:
      from werkzeug.serving import run_simple
      run_simple(config.host, config.port, target_app, use_reloader=False)
    finally:
      print(' * stopping builder threads...')
      build.stop_consumers()

# 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',
                                   DEFAULT_CONFIG_FILES)

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

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

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

    ext_names = ext_master.call('introduce', 'unknown')
    print('Loaded extensions: {}'.format(', '.join(ext_names)))

    ext_master.call('init_first')
    ext_master.call('init_models')
    ext_master.call('init_business')
    ext_master.call('init_filters')
    ext_master.call('init_blueprints')
    ext_master.call('init_container')

    if config.has_option('flask', 'application_root'):
        from werkzeug.serving import run_simple
        from werkzeug.wsgi import DispatcherMiddleware
        app.wsgi_app = DispatcherMiddleware(
            run_simple,
            {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():
        spec=client.getOsisSpecModel(namespace)
        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)
    
        Bootstrap(app)
    
        @app.route('/ui')
        def ui():
            return render_template('ui.html')
    
        # Unfortunately, eve_docs doesn't support CORS (too bad!), so we had to reimplement it ourselves
        @app.route('/docs/spec.json')
        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('0.0.0.0', 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 = '127.0.0.1'
        if port is None:
            server_name = self.config['SERVER_NAME']
            if server_name and ':' in server_name:
                port = int(server_name.rsplit(':', 1)[1])
            else:
                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)

        try:
            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()

        self.logger.info('webui running on %s:%s', hostname, port)
        self.ioloop = tornado.ioloop.IOLoop.current()
        self.ioloop.start()