我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用tornado.wsgi.WSGIContainer()。
def web_server(wiki, port, debug=False): def kill_handler(signal_number, stack_frame): logger.info('\nStopping wiki') sys.exit(1) signal.signal(signal.SIGINT, kill_handler) logger.info('Starting wiki on port {}. Ctrl+C will kill it.'.format(port)) HTTPServer(WSGIContainer(wiki)).listen(port) ioloop = IOLoop.instance() if debug: autoreload.start(ioloop) ioloop.start()
def get_app(self): class HelloHandler(RequestHandler): def get(self): self.write("Hello world!") class PathQuotingHandler(RequestHandler): def get(self, path): self.write(path) # It would be better to run the wsgiref server implementation in # another thread instead of using our own WSGIContainer, but this # fits better in our async testing framework and the wsgiref # validator should keep us honest return WSGIContainer(validator(WSGIApplication([ ("/", HelloHandler), ("/path/(.*)", PathQuotingHandler), ("/typecheck", TypeCheckHandler), ])))
def get_app(self): class HelloHandler(RequestHandler): def get(self): self.write("Hello world!") class PathQuotingHandler(RequestHandler): def get(self, path): self.write(path) # It would be better to run the wsgiref server implementation in # another thread instead of using our own WSGIContainer, but this # fits better in our async testing framework and the wsgiref # validator should keep us honest return WSGIContainer(validator(WSGIApplication([ ("/", HelloHandler), ("/path/(.*)", PathQuotingHandler), ])))
def main(app: Flask, tornado: bool=False) -> None: """Run the Flask application.""" with app.app_context(): db.create_all() print('[OK] Database creation complete.') if tornado: from tornado.wsgi import WSGIContainer from tornado.httpserver import HTTPServer from tornado.ioloop import IOLoop http_server = HTTPServer(WSGIContainer(app)) http_server.listen(int(app.config['INIT_PORT'])) IOLoop.instance().start() else: socketio.run(app, **app.config['INIT'])
def get_app(self): return WSGIContainer(validator(self.wsgi_app))
def test_types(self): headers = {"Cookie": "foo=bar"} response = self.fetch("/typecheck?foo=bar", headers=headers) data = json_decode(response.body) self.assertEqual(data, {}) response = self.fetch("/typecheck", method="POST", body="foo=bar", headers=headers) data = json_decode(response.body) self.assertEqual(data, {}) # This is kind of hacky, but run some of the HTTPServer tests through # WSGIContainer and WSGIApplication to make sure everything survives # repeated disassembly and reassembly.
def get_app(self): return WSGIContainer(validator(WSGIApplication(self.get_handlers())))
def wrap_web_tests_application(): result = {} for cls in web_test.wsgi_safe_tests: class WSGIApplicationWrappedTest(cls): def get_app(self): self.app = WSGIApplication(self.get_handlers(), **self.get_app_kwargs()) return WSGIContainer(validator(self.app)) result["WSGIApplication_" + cls.__name__] = WSGIApplicationWrappedTest return result
def run_tornado_server(app, port): from tornado.wsgi import WSGIContainer from tornado.httpserver import HTTPServer from tornado.ioloop import IOLoop log.info("Activating Tornado server on port %s", port) http_server = HTTPServer(WSGIContainer(app), xheaders=True) http_server.listen(port) IOLoop.instance().start()
def server_main(args=sys.argv): parser = get_server_options_parser() # sys.argv includes the path of invocation as the first index in the list # argparse expects just the parameters if we pass a list into parse_args # so here we get rid of that path parameter and are left with only the args # from the command line if args == sys.argv: args = args[1:] args = parser.parse_args(args) HopliteServerSettings.debug = args.debug app = create_app() logger.info('Starting Hoplite server on port {}'.format(args.port)) http_server = HTTPServer(WSGIContainer(app)) http_server.listen(args.port) ioloop = IOLoop.instance() # This is needed to ensure Ctrl-C kills the server quickly set_ping(ioloop, timedelta(seconds=2)) try: ioloop.start() except: raise finally: # Ensure that the server always closes the socket when it's no longer # in use ioloop.stop()
def main(): define('port', default=9000, type=int, help='Port on which to listen.') parse_command_line() http_server = HTTPServer(WSGIContainer(app)) http_server.listen(options.port) IOLoop.instance().start()
def test_path_quoting(self): response = self.fetch("/path/foo%20bar%C3%A9") self.assertEqual(response.body, u"foo bar\u00e9".encode("utf-8")) # This is kind of hacky, but run some of the HTTPServer tests through # WSGIContainer and WSGIApplication to make sure everything survives # repeated disassembly and reassembly.
def wrap_web_tests(): result = {} for cls in web_test.wsgi_safe_tests: class WSGIWrappedTest(cls): def get_app(self): self.app = WSGIApplication(self.get_handlers(), **self.get_app_kwargs()) return WSGIContainer(validator(self.app)) result["WSGIWrapped_" + cls.__name__] = WSGIWrappedTest return result
def wrap_web_tests_adapter(): result = {} for cls in web_test.wsgi_safe_tests: class WSGIAdapterWrappedTest(cls): def get_app(self): self.app = Application(self.get_handlers(), **self.get_app_kwargs()) return WSGIContainer(validator(WSGIAdapter(self.app))) result["WSGIAdapter_" + cls.__name__] = WSGIAdapterWrappedTest return result
