def cmd_view_pstats(args=None):
"""
Allows calling of view_pstats from a console script.
"""
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('-p', '--port', action='store', dest='port',
default=8009, type=int,
help='port used for web server')
parser.add_argument('--filter', action='store', dest='filter',
default=None,
help='portion of filename used to filter displayed functions.')
parser.add_argument('file', metavar='file', nargs=1,
help='profile file to view.')
options = parser.parse_args(args)
view_pstats(options.file[0], port=options.port, selector=options.filter)
python类web()的实例源码
def _iprof_setup_parser(parser):
if not func_group:
_setup_func_group()
parser.add_argument('-p', '--port', action='store', dest='port',
default=8009, type=int,
help='port used for web server')
parser.add_argument('--no_browser', action='store_true', dest='noshow',
help="Don't pop up a browser to view the data.")
parser.add_argument('-t', '--title', action='store', dest='title',
default='Profile of Method Calls by Instance',
help='Title to be displayed above profiling view.')
parser.add_argument('-g', '--group', action='store', dest='methods',
default='openmdao',
help='Determines which group of methods will be tracked. Current '
'options are: %s and "openmdao" is the default' %
sorted(func_group.keys()))
parser.add_argument('-m', '--maxcalls', action='store', dest='maxcalls',
default=15000, type=int,
help='Maximum number of calls displayed at one time. Default=15000.')
parser.add_argument('file', metavar='file', nargs='+',
help='Raw profile data files or a python file.')
def _iprof_exec(options):
"""
Called from a command line to instance based profile data in a web page.
"""
if options.file[0].endswith('.py'):
if len(options.file) > 1:
print("iprofview can only process a single python file.", file=sys.stderr)
sys.exit(-1)
_iprof_py_file(options)
options.file = ['iprof.0']
if not options.noshow:
app = _Application(options)
app.listen(options.port)
print("starting server on port %d" % options.port)
serve_thread = _startThread(tornado.ioloop.IOLoop.current().start)
launch_thread = _startThread(lambda: _launch_browser(options.port))
while serve_thread.isAlive():
serve_thread.join(timeout=1)
def start(self):
self.logger.info("starting web server listening at https port {0}".format(self.httpsPort))
dir = os.path.dirname(os.path.realpath(__file__))
handlersArgs = dict(iotManager=self.iotManager)
application = [
(r'/(favicon.ico)', tornado.web.StaticFileHandler, {'path': dir + '/img'}),
(r'/static/(.*)', tornado.web.StaticFileHandler, {'path': dir + '/static'}),
(r'/upload/(.*)', handlers.AuthFileHandler, {'path': self.uploadDir}),
(r'/img/(.*)', tornado.web.StaticFileHandler, {'path': dir + '/img'}),
(r'/login', handlers.LoginWebHandler, dict(adminPasswordHash=self.adminPasswordHash)),
(r'/logout', handlers.LogoutWebHandler),
(r'/ws', handlers.WSHandler, handlersArgs),
(r'/device/(.*)', handlers.DeviceWebHandler, handlersArgs),
(r'/rss', handlers.RssWebHandler, handlersArgs),
(r'/history', handlers.HistoryWebHandler, handlersArgs),
(r'/devices', handlers.DevicesWebHandler, handlersArgs),
(r'/video', handlers.VideoWebHandler, dict(localVideoPort=self.localVideoPort)),
(r'/', handlers.HomeWebHandler, handlersArgs),
]
self.logger.info("starting web server listening at http {0} (plain)".format(self.httpPort))
self.httpsApp = tornado.web.Application(application, cookie_secret=os.urandom(32), compiled_template_cache=True)
sslOptions={ "certfile": self.httpsCertFile, "keyfile": self.httpsKeyFile, "ssl_version": ssl.PROTOCOL_TLSv1 }
if self.httpsChainFile:
sslOptions["certfile"] = self.httpsChainFile
self.logger.info("Using certificate file at {0}".format(sslOptions["certfile"]))
self.httpsServer = tornado.httpserver.HTTPServer(self.httpsApp, ssl_options=sslOptions)
self.httpsServer.listen(self.httpsPort)
httpApplication = [
(r'/rss', handlers.RssWebHandler, handlersArgs),
(r'/', handlers.RedirectorHandler, dict(manager = self)),
(r'/(.*)', tornado.web.StaticFileHandler, {'path': dir + '/plain' })
]
self.httpApp = tornado.web.Application(httpApplication)
self.httpServer = tornado.httpserver.HTTPServer(self.httpApp)
self.httpServer.listen(self.httpPort)
tornado.ioloop.IOLoop.current().start()
def __init__(self, application):
if isinstance(application, WSGIApplication):
self.application = lambda request: web.Application.__call__(
application, request)
else:
self.application = application
def write(self, chunk):
"""Writes the given chunk to the output buffer.
To write the output to the network, use the flush() method below.
If the given chunk is a dictionary, we write it as JSON and set
the Content-Type of the response to be ``application/json``.
(if you want to send JSON as a different ``Content-Type``, call
set_header *after* calling write()).
Note that lists are not converted to JSON because of a potential
cross-site security vulnerability. All JSON output should be
wrapped in a dictionary. More details at
http://haacked.com/archive/2009/06/25/json-hijacking.aspx/ and
https://github.com/facebook/tornado/issues/1009
"""
if self._finished:
raise RuntimeError("Cannot write() after finish()")
if not isinstance(chunk, (bytes, unicode_type, dict)):
message = "write() only accepts bytes, unicode, and dict objects"
if isinstance(chunk, list):
message += ". Lists not accepted for security reasons; see http://www.tornadoweb.org/en/stable/web.html#tornado.web.RequestHandler.write"
raise TypeError(message)
if isinstance(chunk, dict):
chunk = escape.json_encode(chunk)
self.set_header("Content-Type", "application/json; charset=UTF-8")
chunk = utf8(chunk)
self._write_buffer.append(chunk)
def get_login_url(self):
"""Override to customize the login URL based on the request.
By default, we use the ``login_url`` application setting.
"""
self.require_setting("login_url", "@tornado.web.authenticated")
return self.application.settings["login_url"]
def __init__(self, application):
if isinstance(application, WSGIApplication):
self.application = lambda request: web.Application.__call__(
application, request)
else:
self.application = application
def write(self, chunk):
"""Writes the given chunk to the output buffer.
To write the output to the network, use the flush() method below.
If the given chunk is a dictionary, we write it as JSON and set
the Content-Type of the response to be ``application/json``.
(if you want to send JSON as a different ``Content-Type``, call
set_header *after* calling write()).
Note that lists are not converted to JSON because of a potential
cross-site security vulnerability. All JSON output should be
wrapped in a dictionary. More details at
http://haacked.com/archive/2009/06/25/json-hijacking.aspx/ and
https://github.com/facebook/tornado/issues/1009
"""
if self._finished:
raise RuntimeError("Cannot write() after finish()")
if not isinstance(chunk, (bytes, unicode_type, dict)):
message = "write() only accepts bytes, unicode, and dict objects"
if isinstance(chunk, list):
message += ". Lists not accepted for security reasons; see http://www.tornadoweb.org/en/stable/web.html#tornado.web.RequestHandler.write"
raise TypeError(message)
if isinstance(chunk, dict):
chunk = escape.json_encode(chunk)
self.set_header("Content-Type", "application/json; charset=UTF-8")
chunk = utf8(chunk)
self._write_buffer.append(chunk)
def __init__(self, application):
if isinstance(application, WSGIApplication):
self.application = lambda request: web.Application.__call__(
application, request)
else:
self.application = application
def write(self, chunk):
"""Writes the given chunk to the output buffer.
To write the output to the network, use the flush() method below.
If the given chunk is a dictionary, we write it as JSON and set
the Content-Type of the response to be ``application/json``.
(if you want to send JSON as a different ``Content-Type``, call
set_header *after* calling write()).
Note that lists are not converted to JSON because of a potential
cross-site security vulnerability. All JSON output should be
wrapped in a dictionary. More details at
http://haacked.com/archive/2009/06/25/json-hijacking.aspx/ and
https://github.com/facebook/tornado/issues/1009
"""
if self._finished:
raise RuntimeError("Cannot write() after finish()")
if not isinstance(chunk, (bytes, unicode_type, dict)):
message = "write() only accepts bytes, unicode, and dict objects"
if isinstance(chunk, list):
message += ". Lists not accepted for security reasons; see http://www.tornadoweb.org/en/stable/web.html#tornado.web.RequestHandler.write"
raise TypeError(message)
if isinstance(chunk, dict):
chunk = escape.json_encode(chunk)
self.set_header("Content-Type", "application/json; charset=UTF-8")
chunk = utf8(chunk)
self._write_buffer.append(chunk)
def get_login_url(self):
"""Override to customize the login URL based on the request.
By default, we use the ``login_url`` application setting.
"""
self.require_setting("login_url", "@tornado.web.authenticated")
return self.application.settings["login_url"]
def main(argv=sys.argv):
"""Main method of the Tenant Webapp Server. This method is encapsulated in a function for packaging to allow it to be
called as a function by an external program."""
config = ConfigParser.SafeConfigParser()
config.read(common.CONFIG_FILE)
webapp_port = config.get('general', 'webapp_port')
logger.info('Starting Tenant WebApp (tornado) on port ' + webapp_port + ', use <Ctrl-C> to stop')
app = tornado.web.Application([
(r"/", MainHandler),
(r"/webapp/.*", WebAppHandler),
(r"/v2/nodes/.*", InstancesHandler),
(r'/static/(.*)', tornado.web.StaticFileHandler, {'path': "static/"}),
])
# WebApp Server TLS
server_context = tenant_templ.get_tls_context()
server_context.check_hostname = False # config.getboolean('general','tls_check_hostnames')
server_context.verify_mode = ssl.CERT_NONE # ssl.CERT_REQUIRED
# Set up server
server = tornado.httpserver.HTTPServer(app,ssl_options=server_context)
server.bind(int(webapp_port), address='0.0.0.0')
server.start(config.getint('cloud_verifier','multiprocessing_pool_num_workers'))
try:
tornado.ioloop.IOLoop.instance().start()
except KeyboardInterrupt:
tornado.ioloop.IOLoop.instance().stop()
def get(self):
ws_id = self.get_argument('id')
Log.objects.filter(id=ws_id).update(is_finished=True)
for ws in WebTerminalHandler.clients:
if ws.id == int(ws_id):
logger.debug("Kill log id %s" % ws_id)
ws.log.save()
ws.close()
logger.debug('Websocket: web terminal client num: %s' % len(WebTerminalHandler.clients))
def main():
from django.core.wsgi import get_wsgi_application
import tornado.wsgi
wsgi_app = get_wsgi_application()
container = tornado.wsgi.WSGIContainer(wsgi_app)
setting = {
'cookie_secret': 'DFksdfsasdfkasdfFKwlwfsdfsa1204mx',
'template_path': os.path.join(os.path.dirname(__file__), 'templates'),
'static_path': os.path.join(os.path.dirname(__file__), 'static'),
'debug': False,
}
tornado_app = tornado.web.Application(
[
(r'/ws/monitor', MonitorHandler),
(r'/ws/terminal', WebTerminalHandler),
(r'/ws/kill', WebTerminalKillHandler),
(r'/ws/exec', ExecHandler),
(r"/static/(.*)", tornado.web.StaticFileHandler,
dict(path=os.path.join(os.path.dirname(__file__), "static"))),
('.*', tornado.web.FallbackHandler, dict(fallback=container)),
], **setting)
server = tornado.httpserver.HTTPServer(tornado_app)
server.listen(options.port, address=IP)
tornado.ioloop.IOLoop.instance().start()
def __init__(self, debug):
root = os.path.dirname(__file__)
static_path = os.path.join(root, 'static')
template_path = os.path.join(root, 'templates')
settings = {
'debug': debug,
'compress_response': True,
'template_path': template_path,
}
# routes
handlers = [
(r'/submit', FormHandler),
(r'/results/?', ResultsSummaryHandler),
(r'/results/([0-9a-fA-F-]+)/?', ResultsHandler),
(r'/tracking/([0-9a-fA-F-]+)/([A-Za-z0-9._-]+)/?', TrackingHandler),
(r'/unsubscribe/([0-9a-fA-F-]+)/?', BlacklistHandler),
(r'/privacy/?', PrivacyHandler),
(r'/', MainHandler),
(r'/(.*)', StaticHandler, {'path': static_path}),
]
# database instance
self.db = db.MailerDatabase(config.DB_PATH)
# rate limiters
self.global_limiter = ratelimit.Bucket(**config.GLOBAL_RATE_LIMIT)
self.ip_limiters = {}
tornado.web.Application.__init__(self, handlers, **settings)
def get(self):
# get user info
summary = self.application.db.get_user_summary()
# categorize by platform and client
data = {}
for row in summary:
platform_data = data.setdefault(row['platform'], {})
client_data = platform_data.setdefault(row['client'], [])
client_data.append({'id': row['id'], 'timestamp': row['timestamp']})
# sort data
sorted_data = []
for platform, name in (('web', 'Web'), ('desktop', 'Desktop'), ('mobile', 'Mobile')):
if platform not in data:
continue
sorted_clients = []
platform_data = data[platform]
for client in sorted(platform_data, key=lambda s: s.lower()):
client_data = list(reversed(platform_data[client]))
sorted_clients.append((client, client_data))
sorted_data.append((name, sorted_clients))
# render page
self.render('summary.html', data=sorted_data)
def __init__(self, *args, **kwargs):
tornado.web.RequestHandler.__init__(self, *args, **kwargs)
# TODO: consider storing original payload for hashing
def run():
logging.basicConfig(format="%(asctime)s %(levelname)s - %(message)s", level=logging.DEBUG)
log = logging.getLogger("txn-service")
log.info("Setting up argparse")
parser = argparse.ArgumentParser(description='Process some integers.', prog='python -m blockchain')
parser.add_argument('-p', '--port', default=8000)
parser.add_argument('--debug', default=True, action="store_true")
parser.add_argument('--private-key', dest="private_key", required=True, help="ECDSA private key for signing")
parser.add_argument('--public-key', dest="public_key", required=True, help="ECDSA private key for signing")
log.info("Parsing arguments")
args = parser.parse_args()
hdlrs = [
(r"^/transaction$", TransactionHandler),
(r"^/transaction/(.*)", TransactionHandler),
]
log.info("Creating new tornado.web.Application")
application = TransactionService(hdlrs,
log = log,
**vars(args))
log.info("Starting transaction service on port %s" % args.port)
application.listen(args.port)
tornado.ioloop.IOLoop.current().start()
def __init__(self, *args, **kwargs):
self.log = kwargs["log"]
del kwargs["log"]
# constructor of base class
tornado.web.Application.__init__(self, *args, **kwargs)