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
python类DebuggedApplication()的实例源码
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()
