python类web()的实例源码

def __init__(self):
        handlers = [
            (r'/monitor', MonitorHandler),
            (r'/terminal', WebTerminalHandler),
            (r'/kill', WebTerminalKillHandler),
            (r'/exec', ExecHandler),
        ]

        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.web.Application.__init__(self, handlers, **setting)

def __init__(self, *args, **kwargs):

        self.private_key = kwargs["private_key"]
        del kwargs["private_key"]

        with open(self.private_key, 'r') as key:
            self.private_key = key.read()

        self.public_key = kwargs["public_key"]
        del kwargs["public_key"]

        with open(self.public_key, 'r') as key:
            self.public_key = key.read()

        self.log = kwargs["log"]
        del kwargs["log"]

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

def __init__(self, *args, **kwargs):
        tornado.web.RequestHandler.__init__(self, *args, **kwargs)
        self.query_fields = {
            'block_id': None,
            'transaction_type': None,
            'create_ts': None,
            'transaction_ts': None,
            'business_unit': None,
            'family_of_business': None,
            'line_of_business': None,
            'signature': None,
            'status': None,
            'actor': None,
            'entity': None,
            'owner': None
        }

def post(self):
        id = self.get_argument("id")
        name = self.get_argument("name")
        content = self.get_argument("content")
        memo = self.get_argument("memo")
        web_resource = ResWeb(
                id=id,
                name=name,
                content=content,
                memo=memo
        )
        try:
            self.session.merge(web_resource)
            self.session.commit()
            self.redirect('/res/web')
        except:
            err = "??????"
            web_resource = self.session.query(ResWeb).filter(ResWeb.id == id).first()
            self.render("res_web_detail.html", auth_user=self.current_user, web_resource=web_resource, err=err)


# ????--??

def post(self):
        web_name = self.get_argument("name")
        web_content = self.get_argument("content")
        memo = self.get_argument("memo")
        web_resource = ResWeb(
                name=web_name,
                content=web_content,
                memo=memo,
                data=time.strftime("%Y-%m-%d"),
                company_id=self.get_secure_cookie('company_id')
        )
        try:
            self.session.add(web_resource)
            self.session.commit()
            self.redirect('/res/web')
        except:
            action = '/res/web/add'
            err = "??????"
            web_resource = ''
            self.render("res_web_detail.html", auth_user=self.current_user, action=action, web_resource=web_resource, err=err)


# ????-??

def make_web_app():
    logging.config.dictConfig(config.LOGGING_CONFIG)

    settings = {
        'debug': constants.DEBUG,
        'template_path': os.path.join(
            os.path.dirname(__file__), "web", "templates"
        ),
        'static_path': os.path.join(
            os.path.dirname(__file__), 'web', 'static'
        ),
        'default_handler_class ': BaseHandler
    }

    app = tornado.web.Application(url_handlers, **settings)
    return app

def __init__(self):
        handlers = [
            (r'/monitor', MonitorHandler),
            (r'/terminal', WebTerminalHandler),
            (r'/kill', WebTerminalKillHandler),
            (r'/exec', ExecHandler),
        ]

        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.web.Application.__init__(self, handlers, **setting)

def check_site_access(self):
        userId = self.get_id_from_cookie() or None
        siteRole = self.get_id_from_cookie(role=True, for_site=Options['site_name'])

        # Check if pre-authorized for site access
        if Options['site_name']:
            # Check if site is explicitly authorized (user has global admin/grader role, or has explicit site listed, including guest users)
            preAuthorized = siteRole is not None
        else:
            # Single site: check if userid is special (admin/grader/guest)
            preAuthorized = Global.userRoles.is_special_user(userId)

        if preAuthorized:
            return

        if Global.login_domain and '@' in userId:
            # External user
            raise tornado.web.HTTPError(403, log_message='CUSTOM:User %s not pre-authorized to access site' % userId)

        # Check if userId appears in roster
        if sdproxy.getSheet(sdproxy.ROSTER_SHEET):
            if not sdproxy.lookupRoster('id', userId):
                raise tornado.web.HTTPError(403, log_message='CUSTOM:Userid %s not found in roster' % userId)
        elif sdproxy.Settings['require_roster']:
                raise tornado.web.HTTPError(403, log_message='CUSTOM:No roster available for site')

def imageUpload(self, sessionName, imageFile, fname, fbody, autonumber=None):
        if Options['debug']:
            print >> sys.stderr, 'ActionHandler:imageUpload', sessionName, imageFile, fname, autonumber, len(fbody)
        if not self.previewActive():
            raise tornado.web.HTTPError(404, log_message='CUSTOM:Not previewing session')
        if not imageFile:
            imgName = re.sub(r'[^\w,.+-]', '', fname.strip().replace(' ','_'))
            if imgName and not autonumber:
                imageFile = imgName
            else:
                imageFile = (md2md.IMAGE_FMT % self.previewState['new_image_number']) + os.path.splitext(fname)[1].lower()
            self.previewState['new_image_number'] += 1
        if not self.previewState['image_zipfile']:
            self.previewState['image_zipbytes'] = io.BytesIO()
            self.previewState['image_zipfile'] = zipfile.ZipFile(self.previewState['image_zipbytes'], 'a')
        imagePath = sessionName+'_images/' + imageFile
        self.previewState['image_zipfile'].writestr(imagePath, fbody)
        self.previewState['image_paths'][imageFile] = imagePath
        self.previewState['modimages'] = 'append'

        self.set_header('Content-Type', 'application/json')
        self.write( json.dumps( {'result': 'success', 'imageFile': imageFile} ) )

def main(config, output):
    try:
        import webbrowser
    except ImportError:
        print('Go to http://{your device IP}:3000 to start')
        login(config, output)
        return

    import threading
    webserver = threading.Thread(target=login, args=(config, output))
    webserver.daemon = True
    webserver.start()
    print("A web page should is opened. If not, go to http://127.0.0.1:3000 to start")
    webbrowser.open('http://127.0.0.1:3000')

    while webserver.is_alive():
        try:
            time.sleep(1)
        except KeyboardInterrupt:
            break

def load(pglot, user_config):
    ''' setup the http server '''
    # pylint: disable=global-statement, unused-argument
    global SERVER, _THREAD

    # set configuration
    config = DEFAULT_CONFIG
    config.update(user_config)
    set_config(config)

    # create server
    SERVER = tornado.web.Application([], {})
    SERVER.listen(PORT)


    # run server on a thread
    _THREAD = threading.Thread(target=run_server)
    _THREAD.daemon = True
    _THREAD.start()
    _LOGGER.info('Started HTTP server on port %d', PORT)

    _LOGGER.info('Loaded HTTP element')

def get(self):
        '''
            get????
        '''

        echo_str = self.get_argument('echostr', default='_ARG_DEFAULT')
        # ????????
        if echo_str.strip() != '_ARG_DEFAULT':
            self.write(echo_str)
        else:
            htmp_file_name = self.get_argument(request_params.KEY_REQ_GET_HTML_FILE_NAME, default='_ARG_DEFAULT')
            # ??????
            if htmp_file_name == '_ARG_DEFAULT':
                self.write('this is a tornador web server!')
            else:
                # ????html????
                get_process_rst = yield self.get_thread(htmp_file_name)
                self.render(get_process_rst)
                # self.finish()

def __init__(self):
        handlers = [
            (r'/monitor', MonitorHandler),
            (r'/terminal', WebTerminalHandler),
            (r'/kill', WebTerminalKillHandler),
            (r'/exec', ExecHandler),
        ]

        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.web.Application.__init__(self, handlers, **setting)

def exception_control(func):
    ''' ???????
    '''
    @functools.wraps(func)
    def wrapper(self):
        ''' ????
        '''
        try:
            code, msg, body = E_SUCC, "OK", func(self)
        except (MissingArgumentError, AssertionError) as ex:
            code, msg, body = E_PARAM, str(ex), None
        except tornado.web.HTTPError:
            raise
        except Exception as ex:
            code, msg, body = E_INTER, str(ex), None
            log_msg = self.request.uri \
                if self.request.files else \
                "%s %s" % (self.request.uri, self.request.body)
            logging.error(log_msg, exc_info=True)
        self.send_json(body, code, msg)
    return wrapper

def process_module(self, module):
        '''
        ??????
        '''
        module = module or ''
        if self.module_prefix:
            module = '%s/%s' % (self.module_prefix, module)
        module = '__'.join([i for i in module.split('/') if i])
        method = getattr(self, module or 'index', None)
        if method and module not in ('get', 'post'):
            try:
                result = method()
                if result:
                    self.send_json(result)
            except Exception as ex:
                logging.error('%s\n%s\n', self.request, str(ex), exc_info=True)
                self.send_json(None, E_INTER, str(ex))
        else:
            raise tornado.web.HTTPError(404)

def __init__(self,
                 ):

        handlers = [(r'/',handle_front,),
                    (r'/ping',handle_ping,),
                    (r'/stats',handle_stats,),
                    (r'/stats_annotation',handle_stats_annotation,),
                    (r'/search',handle_search,),
                    (r'/list_facets',handle_list_facets),
                    (r'/get_embed_url',handle_get_embed_url,),
                    (r'/record_relevance',handle_record_relevance,),
                    (r'/random_query',handle_random_query,),
                    #(r'.*', handle_notfound,),
                    ]

        settings = {'template_path':join(dirname(__file__), 'templates_mc'),
                    'static_path':join(dirname(__file__), 'static_mc'),
                    'xsrf_cookies':False,
                    }

        tornado.web.Application.__init__(self, handlers, **settings)

        self.INDEX_NAME = mc_config.MC_INDEX_NAME
        self.DOC_TYPE = mc_config.MC_DOC_TYPE

def web(port = 23456,
        via_cli = False,
        ):
    """
    Bind Tornado server to specified port.    
    """

    print ('BINDING',port)

    try:
        tornado.options.parse_command_line()
        http_server = HTTPServer(Application(),
                                 xheaders=True,
                                 )
        http_server.bind(port)
        http_server.start(16) # Forks multiple sub-processes
        tornado.ioloop.IOLoop.instance().set_blocking_log_threshold(0.5)
        IOLoop.instance().start()

    except KeyboardInterrupt:
        print 'Exit'

    print ('WEB_STARTED')

def validate_absolute_path(self, root, absolute_path):
        """
        An override of
        :meth:`tornado.web.StaticFileHandler.validate_absolute_path`;

        Validate and returns the given *absolute_path* using `pkg_resources`
        if ``self.use_pkg`` is set otherwise performs a normal filesystem
        validation.
        """
        # We have to generate the real absolute path in this method since the
        # Tornado devs--for whatever reason--decided that get_absolute_path()
        # must be a classmethod (we need access to self.use_pkg).
        if self.use_pkg:
            if not resource_exists(self.use_pkg, absolute_path):
                raise HTTPError(404)
            return resource_filename(self.use_pkg, absolute_path)
        return super(
            StaticHandler, self).validate_absolute_path(root, absolute_path)

def add_handler(self, pattern, handler, **kwargs):
        """
        Adds the given *handler* (`tornado.web.RequestHandler`) to the Tornado
        Application (`self.ws.application`) to handle URLs matching *pattern*.
        If given, *kwargs* will be added to the `tornado.web.URLSpec` when the
        complete handler is assembled.

        .. note::

            If the *pattern* does not start with the configured `url_prefix` it
            will be automatically prepended.
        """
        logging.debug("Adding handler: (%s, %s)" % (pattern, handler))
        url_prefix = self.ws.settings['url_prefix']
        if not pattern.startswith(url_prefix):
            if pattern.startswith('/'):
                # Get rid of the / (it will be in the url_prefix)
                pattern = pattern.lstrip('/')
        spec = tornado.web.URLSpec(pattern, handler, kwargs)
        # Why the Tornado devs didn't give us a simple way to do this is beyond
        # me.
        self.ws.application.handlers[0][1].append(spec)

def on_message(self, message):
        logging.info('from web client: {}'.format(message))
        msg = tornado.escape.json_decode(tornado.escape.to_basestring(message))

        cmd = msg.get('cmd')
        if cmd == 'close':
            if 'data' in msg and 'eid' in msg:
                logging.info('closing window {}'.format(msg['data']))
                self.state[msg['eid']]['jsons'].pop(msg['data'], None)

        elif cmd == 'save':
            # save localStorage window metadata
            if 'data' in msg and 'eid' in msg:
                msg['eid'] = escape_eid(msg['eid'])
                self.state[msg['eid']] = copy.deepcopy(self.state[msg['prev_eid']])
                self.state[msg['eid']]['reload'] = msg['data']
                self.eid = msg['eid']
                serialize_env(self.state, [self.eid])

def on_message(self, message):
            global clientStatus
            global robotStatus
            if(message == 'webApp'):
                print('[Server] Connected to web app')
                global apps
                apps.append(self)
                self.isApp = True
            elif(message == 'client'):
                print('[Server] Connected to client')
                global clients
                clientStatus = True
                clients.append(self)
                self.isClient = True
                for app in apps:
                    app.write_message('client,connected')
            else:
                global commandHandler
                commandHandler(message)
            if(message == 'webApp' or message == 'client'):
                global clientCount
                self.write_message('id,' + str(clientCount));
                clientCount += 1;
                self.write_message('cozmo,' + ('connected' if robotStatus else 'waiting'))
                self.write_message('client,' + ('connected' if clientStatus else 'waiting'))

def start(self):
        application = tornado.web.Application([
        (r'/ws', Server.WSHandler),
        (r'/', Server.AppHandler),
        (r'/blockly', Server.BlocklyHandler),
        (r'/blockly/code/(.*)', tornado.web.StaticFileHandler, {'path': 'blockly'}),
        (r'/stream', Server.StreamHandler),
        (r'/stream/(.*)', tornado.web.StaticFileHandler, {'path': 'Web App/stream'}),
        (r'blockly/code/msg/js/(en.js)', tornado.web.StaticFileHandler, {'path': 'blockly/msg/js'}),
        (r'/(theme\.css)', tornado.web.StaticFileHandler, {'path': 'Web App'}),
        (r'/(connector\.js)', tornado.web.StaticFileHandler, {'path': 'Web App'}),
        (r'/image/(scratch\.png)', tornado.web.StaticFileHandler, {'path': 'Web App/images'}),
        (r'/image/(blockly\.png)', tornado.web.StaticFileHandler, {'path': 'Web App/images'}),
        (r'/image/(cozmo\.png)', tornado.web.StaticFileHandler, {'path': 'Web App/images'}),
        ])
        print('[Server] Starting server...')
        application.listen(9090)
        print("[Server] Server ready at: localhost:9090")
        print("[Server] Websockets ready at: localhost:9090/ws")
        webbrowser.open("http://localhost:9090")
        print("[Server] Web browser openned to: http://localhost:9090")
        tornado.ioloop.IOLoop.instance().start()
        print('[Server] Server stopped')

def __init__(self, figure):
        self.figure = figure
        self.manager = new_figure_manager_given_figure(
            id(figure), figure)

        super(MyApplication, self).__init__([
            # Static files for the CSS and JS
            (r'/_static/(.*)',
             tornado.web.StaticFileHandler,
             {'path': FigureManagerWebAgg.get_static_file_path()}),

            # The page that contains all of the pieces
            ('/', self.MainPage),

            ('/mpl.js', self.MplJs),

            # Sends images and events to the browser, and receives
            # events from the browser
            ('/ws', self.WebSocket),

            # Handles the downloading (i.e., saving) of static images
            (r'/download.([a-z0-9.]+)', self.Download),
        ], debug=True)

def get(self, latlon=None):
        if latlon == None:
            raise tornado.web.HTTPError(500) 
        latlon = latlon.split(',')
        lat = float(latlon[0])
        lon = float(latlon[1])
        stations = [];
        print("tide station search near: %f,%f" % (lat,lon))
        for id,t in __tcd_stations__.items():
            km = self.measureDistance(lat, lon, t.latitude, t.longitude)
            if km < 2000:
                station = { "id":t.record_number, "name":t.name, "distance":km }
                stations.append(station)

        stations = sorted(stations, key=lambda station: station["distance"])
        if len(stations) > 10: stations = stations[0:9]
        returnData = {"stations":stations}
        json = tornado.escape.json_encode(returnData)

        self.write(json)
        self.finish()

def __init__(self, port, job_manager, mongo_uri):
        self.job_manager = job_manager
        self.mongo_uri = mongo_uri
        self.static_path = os.path.join(os.path.dirname(__file__), "static")

        self.routes = Application(
        [
            (r"/", MainRequestHandler),
            (r"/worker/(.*)/job", JobHandler, {"job_manager": job_manager}),
            (r"/stats", StatsHandler, {"mongo_uri": self.mongo_uri}),
            (r"/stats/(.*)", StatsHandler, {"mongo_uri": self.mongo_uri}),
            (r"/data/(.*)", tornado.web.StaticFileHandler, {'path': self.static_path})
            ],
        template_path=os.path.join(os.path.dirname(__file__), "templates"),
        static_path= self.static_path)

        self.port = port

def handle_response(self, response):
        if response.error and not isinstance(response.error, tornado.web.HTTPError):
            raise tornado.web.HTTPError(500)

        filename = None

        self.set_status(response.code)
        for name in ("Date", "Cache-Control", "Server", "Content-Type", "Location", "Expires", "ETag"):
            value = response.headers.get(name)
            if value:
                self.set_header(name, value)

                if name == "Content-Type":
                    filename = self.get_filename(value)

        if self._as_attachment:
            if filename is not None:
                self.set_header("Content-Disposition", "attachment; filename=%s" % filename)
            else:
                self.set_header("Content-Disposition", "attachment")

        if response.body:
            self.write(response.body)
        self.finish()

def start_app(cls, handlers=None, settings=None):
        """
            Initialize the Tornado web application with given handlers and settings.

            :param handlers: Handlers (routes) for Tornado
            :param settings: Settings for Tornado
            :type handlers: list
            :type settings: dict
            :return: None
        """

        if not handlers:
            handlers = []

        if not settings:
            settings = {}

        # Not `handlers += cls.handlers` because the wildcard handler should be the last value in handlers
        # list. See http://www.tornadoweb.org/en/stable/_modules/tornado/web.html#Application.add_handlers
        handlers = cls.handlers + handlers

        cls.app = tornado.web.Application(handlers, **settings)

def init(port):
    template_path = os.path.join(
        os.path.dirname(os.path.realpath(__file__)),
        "..", "assets", "templates")
    static_files_path = os.path.join(
        os.path.dirname(os.path.realpath(__file__)),
        "..", "assets", "web")
    favicon_path = os.path.join(
        os.path.dirname(os.path.realpath(__file__)),
        "..", "assets")
    app = tornado.web.Application([
        tornado.web.url(r"/", MainHandler, name="main"),
        tornado.web.url(r"/login", LoginHandler, name="login"),
        tornado.web.url(r"/questions/new", NewQuestionHandler, name="new_question"),
        tornado.web.url(r"/questions/([^/]+)", QuestionHandler, name="question"),
        tornado.web.url(r"/ws/notify", NotificationWebSocketHandler, name="notify_websocket"),
        tornado.web.url(r"/ws/(.*)", QuestionWebSocketHandler, name="question_websocket"),
        tornado.web.url(r"/assets/(.*)", tornado.web.StaticFileHandler, {"path": static_files_path}, name="static"),
        tornado.web.url(r"/(favicon\.ico)", tornado.web.StaticFileHandler, {"path": favicon_path}, name="favicon")
    ], template_path=template_path, debug=True)
    server = tornado.httpserver.HTTPServer(app)
    server.listen(port)

def asyncGet(self):
        try:
            args = {}
            #if "stream" in self.request.arguments:
            #   args["stream"] = self.get_argument("stream")
            #if "action" in self.request.arguments:
            #   args["action"] = self.get_argument("action")
            #if "time" in self.request.arguments:
            #   args["time"] = self.get_argument("time")

            # Pass all arguments otherwise it doesn't work
            for key, _ in self.request.arguments.items():
                args[key] = self.get_argument(key)

            if args["action"].lower() == "put":
                self.write("nope")
                return

            response = requests.get("https://osu.ppy.sh/web/check-updates.php?{}".format(urlencode(args)))
            self.write(response.text)
        except Exception as e:
            log.error("check-updates failed: {}".format(e))
            self.write("")

def __init__(self):
        handlers = [
            (r'/monitor', MonitorHandler),
            (r'/terminal', WebTerminalHandler),
            (r'/kill', WebTerminalKillHandler),
            (r'/exec', ExecHandler),
        ]

        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.web.Application.__init__(self, handlers, **setting)