Python werkzeug.debug 模块，DebuggedApplication() 实例源码

def load(self, proxy_mode=False, rules=None, newrelic_agent=None):
        _logger.info("Loading Odoo WSGI application")
        self.load_server_wide_modules()

        application = odoo_application

        if config['debug_mode']:
            application = DebuggedApplication(application, evalex=True)
            _logger.warning("Debugger enabled, do not use in production")

        if newrelic_agent:
            application = newrelic_agent.WSGIApplicationWrapper(application)
            _logger.info("New Relic enabled")

        if rules and rules.has_rules():
            application = rules(application)
            _logger.info("Rewrites enabled")

        if proxy_mode:
            application = ProxyFix(application)
            _logger.info("Proxy mode enabled")

        return application

def register(self):
        """
        Register the service provider
        """

        config_debug = self.app.config('app.debug', None)

        # Correct the debug mode
        if 'FLASK_DEBUG' not in os.environ:
            if config_debug is not None:
                self.app.debug = config_debug
            elif self.app.is_development():
                self.app.debug = True

        # Debug Application for Google App Engine as this is not loaded by default
        if self.app.debug and self.app.is_gae():
            self.app.wsgi_app = DebuggedApplication(self.app.wsgi_app)

        # Testing environment
        if self.app.is_testing():
            self.app.testing = True

def runserver(host="127.0.0.1", port=8080):
    """Run a gevent-based WSGI server."""
    port = int(port)

    wrapped_app = app
    if app.config.get("DEBUG", True):
        #it is necessary to do this for the debug to work, since the classic
        # flask debugger does not operate when running the app from
        # a WSGIServer() call.
        wrapped_app = DebuggedApplication(app)

    server = WSGIServer(listener=(host, port), application=wrapped_app,)

    def serve():
        print(" * Running on http://%s:%d/" % (host, port))
        server.serve_forever()

    if app.debug:
        # the watchdog reloader (with watchdog==0.8.3) appears to hang on
        # Windows 8.1, so we're using stat instead
        run_with_reloader(
            serve, reloader_type="stat" if sys.platform == "win32" else "auto")
    else:
        serve()

def run(self, host='localhost', port=5000, **options):
        """Runs the application on a local development server"""
        # from werkzeug import run_simple
        from gevent.wsgi import WSGIServer
        from werkzeug.debug import DebuggedApplication
        if 'debug' in options:
            self.debug = options.pop('debug')
        # if self.static_path is not None:
        #     options['static_files'] = {
        #         self.static_path:   (self.package_name, 'static')
        #     }
        # options.setdefault('use_reloader', self.debug)
        # options.setdefault('use_debugger', self.debug)
        # return run_simple(host, port, self, **options)
        app = DebuggedApplication(self, evalex=False)
        WSGIServer((host, port), app).serve_forever()

def run_debug(self, host, port):
        """
        ??leics??
            ????: debug
        """
        app = DebuggedApplication(self, evalex=True)
        print "\033[31m{Debug}\033[0m [leics] is running on [http://%s:%d]" % (host, port)
        WSGIServer((host, port), app).serve_forever()

def start(self):
        handler = self.handleRequest

        if config.debug:
            # Auto reload UiRequest on change
            from Debug import DebugReloader
            DebugReloader(self.reload)

            # Werkzeug Debugger
            try:
                from werkzeug.debug import DebuggedApplication
                handler = DebuggedApplication(self.handleRequest, evalex=True)
            except Exception, err:
                self.log.info("%s: For debugging please download Werkzeug (http://werkzeug.pocoo.org/)" % err)
                from Debug import DebugReloader
        self.log.write = lambda msg: self.log.debug(msg.strip())  # For Wsgi access.log
        self.log.info("--------------------------------------")
        self.log.info("Web interface: http://%s:%s/" % (config.ui_ip, config.ui_port))
        self.log.info("--------------------------------------")

        if config.open_browser:
            logging.info("Opening browser: %s...", config.open_browser)
            import webbrowser
            if config.open_browser == "default_browser":
                browser = webbrowser.get()
            else:
                browser = webbrowser.get(config.open_browser)
            browser.open("http://%s:%s" % (config.ui_ip if config.ui_ip != "*" else "127.0.0.1", config.ui_port), new=2)

        self.server = WSGIServer((self.ip.replace("*", ""), self.port), handler, handler_class=UiWSGIHandler, log=self.log)
        self.server.sockets = {}
        try:
            self.server.serve_forever()
        except Exception, err:
            self.log.error("Web interface bind error, must be running already, exiting.... %s" % err)
            sys.modules["main"].file_server.stop()
        self.log.debug("Stopped.")

def run_gunicorn_app(host, port, debug, **settings):
    """Serve Flask application using Gunicorn.

    The Flask application and respective resources and endpoints should
    defined in `app.py` in this same directory.
    """

    logging.basicConfig(level='DEBUG' if debug else 'INFO')

    # Set a global flag that indicates that we were invoked from the
    # command line interface provided server command.  This is detected
    # by Flask.run to make the call into a no-op.  This is necessary to
    # avoid ugly errors when the script that is loaded here also attempts
    # to start a server.
    os.environ['FLASK_RUN_FROM_CLI_SERVER'] = '1'

    settings['bind'] = '{}:{}'.format(host, port)

    if debug:
        app.jinja_env.auto_reload = True
        app.config['TEMPLATES_AUTO_RELOAD'] = True
        settings.update({'loglevel': 'debug',
                         'reload': True,
                         'threads': 1,
                         'workers': 1,
                         'worker_class': 'sync'})
        app.wsgi_app = DebuggedApplication(app.wsgi_app, True)
        logging.info(" * Launching in Debug mode.")
        logging.info(" * Serving application using a single worker.")
    else:
        logging.info(" * Launching in Production Mode.")
        logging.info(" * Serving application with {} worker(s)."
                     .format(settings["workers"]))

    server = GunicornApp(app, settings=settings)
    server.run()

def run_server():
        http_server = WSGIServer(('', 5000), DebuggedApplication(application))
        http_server.serve_forever()

def test_import_string(self):
        import cgi
        from werkzeug.debug import DebuggedApplication
        self.assert_is(utils.import_string('cgi.escape'), cgi.escape)
        self.assert_is(utils.import_string(u'cgi.escape'), cgi.escape)
        self.assert_is(utils.import_string('cgi:escape'), cgi.escape)
        self.assert_is_none(utils.import_string('XXXXXXXXXXXX', True))
        self.assert_is_none(utils.import_string('cgi.XXXXXXXXXXXX', True))
        self.assert_is(utils.import_string(u'cgi.escape'), cgi.escape)
        self.assert_is(utils.import_string(u'werkzeug.debug.DebuggedApplication'), DebuggedApplication)
        self.assert_raises(ImportError, utils.import_string, 'XXXXXXXXXXXXXXXX')
        self.assert_raises(ImportError, utils.import_string, 'cgi.XXXXXXXXXX')

def __init__(self, *args, **kwargs):
        super(DebugApplication, self).__init__(*args, **kwargs)
        self.debug_app = DebuggedApplication(self, evalex=True)
        self.debug_container = tornado.wsgi.WSGIContainer(self.debug_app)

def gevent_run(app, address, port, debugger=False, reloader=False):
    from gevent.wsgi import WSGIServer
    from werkzeug.debug import DebuggedApplication
    import gevent.monkey
    gevent.monkey.patch_all()

    run_app = app
    if debugger:
        run_app = DebuggedApplication(app)

    def run_server():
        import logging
        from gevent import version_info

        logger = logging.getLogger('pydirl')
        logger.info('Listening on http://{}:{}/'.format(address, port))
        server_params = dict()
        # starting from gevent version 1.1b1 we can pass custom logger to gevent
        if version_info[:2] >= (1, 1):
            server_params['log'] = logger
        http_server = WSGIServer((address, port), run_app, **server_params)
        http_server.serve_forever()

    if reloader:
        from werkzeug._reloader import run_with_reloader
        run_with_reloader(run_server)
    else:
        run_server()

def run():
        global application
        app.debug = True
        application = DebuggedApplication(application, evalex=True)
        server = WSGIServer(('localhost', PORT), application, handler_class=WebSocketHandler)
        server.serve_forever()

def __call__(self, app, host, port, use_debugger, use_reloader, threaded, processes, passthrough_errors):
        if use_debugger is None:
            use_debugger = app.debug
            if use_debugger is None:
                use_debugger = True
                if sys.stderr.isatty():
                    print("Debugging is on. DANGER: Do not allow random users to connect to this server.",
                          file=sys.stderr)
        if use_reloader is None:
            use_reloader = app.debug

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

        def run():
            from gevent.wsgi import WSGIServer
            gws = WSGIServer((host, port), app)
            gws.base_env['wsgi.multithread'] = threaded
            gws.base_env['wsgi.multiprocess'] = processes > 0
            gws.serve_forever()

        if use_reloader:
            from werkzeug.serving import run_with_reloader
            run_with_reloader(run)
        else:
            run()

def firing(self, host='localhost', port=8090, **options):
        # run coroutx app on gevent wsgiserver
        from gevent.wsgi import WSGIServer
        if 'debug' in options:
            self.debug = options.pop('debug')
            if self.debug:
                from werkzeug.debug import DebuggedApplication
                app = DebuggedApplication(self, evalex=True)
        else:
            _app = self
        print """coroutx app running on {%s => %s}\n""" % (host, port)
        WSGIServer((host, port), _app).serve_forever()

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