python类web()的实例源码

iprof_app.py 文件源码 项目:OpenMDAO 作者: OpenMDAO 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
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)
iprofile_app.py 文件源码 项目:OpenMDAO 作者: OpenMDAO 项目源码 文件源码 阅读 20 收藏 0 点赞 0 评论 0
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.')
iprofile_app.py 文件源码 项目:OpenMDAO 作者: OpenMDAO 项目源码 文件源码 阅读 17 收藏 0 点赞 0 评论 0
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)
web.py 文件源码 项目:IotCenter 作者: panjanek 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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()
wsgi.py 文件源码 项目:noc-orchestrator 作者: DirceuSilvaLabs 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
def __init__(self, application):
        if isinstance(application, WSGIApplication):
            self.application = lambda request: web.Application.__call__(
                application, request)
        else:
            self.application = application
web.py 文件源码 项目:noc-orchestrator 作者: DirceuSilvaLabs 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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)
web.py 文件源码 项目:noc-orchestrator 作者: DirceuSilvaLabs 项目源码 文件源码 阅读 23 收藏 0 点赞 0 评论 0
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"]
wsgi.py 文件源码 项目:noc-orchestrator 作者: DirceuSilvaLabs 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
def __init__(self, application):
        if isinstance(application, WSGIApplication):
            self.application = lambda request: web.Application.__call__(
                application, request)
        else:
            self.application = application
web.py 文件源码 项目:noc-orchestrator 作者: DirceuSilvaLabs 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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)
wsgi.py 文件源码 项目:noc-orchestrator 作者: DirceuSilvaLabs 项目源码 文件源码 阅读 27 收藏 0 点赞 0 评论 0
def __init__(self, application):
        if isinstance(application, WSGIApplication):
            self.application = lambda request: web.Application.__call__(
                application, request)
        else:
            self.application = application
web.py 文件源码 项目:noc-orchestrator 作者: DirceuSilvaLabs 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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)
web.py 文件源码 项目:noc-orchestrator 作者: DirceuSilvaLabs 项目源码 文件源码 阅读 21 收藏 0 点赞 0 评论 0
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"]
tenant_webapp.py 文件源码 项目:python-keylime 作者: mit-ll 项目源码 文件源码 阅读 24 收藏 0 点赞 0 评论 0
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()
run_server.py 文件源码 项目:geekcloud 作者: Mr-Linus 项目源码 文件源码 阅读 29 收藏 0 点赞 0 评论 0
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))
run_server.py 文件源码 项目:geekcloud 作者: Mr-Linus 项目源码 文件源码 阅读 26 收藏 0 点赞 0 评论 0
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()
webserver.py 文件源码 项目:email-tracking-tester 作者: itdelatrisu 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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)
webserver.py 文件源码 项目:email-tracking-tester 作者: itdelatrisu 项目源码 文件源码 阅读 22 收藏 0 点赞 0 评论 0
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)
transaction_svc.py 文件源码 项目:dragonchain 作者: dragonchain 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def __init__(self, *args, **kwargs):
        tornado.web.RequestHandler.__init__(self, *args, **kwargs)

    # TODO: consider storing original payload for hashing
transaction_svc.py 文件源码 项目:dragonchain 作者: dragonchain 项目源码 文件源码 阅读 18 收藏 0 点赞 0 评论 0
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()
query_svc.py 文件源码 项目:dragonchain 作者: dragonchain 项目源码 文件源码 阅读 19 收藏 0 点赞 0 评论 0
def __init__(self, *args, **kwargs):
        self.log = kwargs["log"]
        del kwargs["log"]

        # constructor of base class
        tornado.web.Application.__init__(self, *args, **kwargs)


问题


面经


文章

微信
公众号

扫码关注公众号