我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用gevent.wsgi.WSGIServer()。
def start_server(): log.setLevel(logging.INFO) logging.getLogger('PokeAlarm').setLevel(logging.INFO) logging.getLogger('requests').setLevel(logging.WARNING) logging.getLogger('pyswgi').setLevel(logging.WARNING) logging.getLogger('connectionpool').setLevel(logging.WARNING) logging.getLogger('gipc').setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) parse_settings(os.path.abspath(os.path.dirname(__file__))) # Start Webhook Manager in a Thread spawn(manage_webhook_data, data_queue) # Start up Server log.info("PokeAlarm is listening for webhooks on: http://{}:{}".format(config['HOST'], config['PORT'])) server = wsgi.WSGIServer((config['HOST'], config['PORT']), app, log=logging.getLogger('pyswgi')) server.serve_forever() ################################################## CONFIG UTILITIES ###################################################
def run(self, app): # pragma: no cover from wsgiref.simple_server import WSGIRequestHandler, WSGIServer from wsgiref.simple_server import make_server import socket class FixedHandler(WSGIRequestHandler): def address_string(self): # Prevent reverse DNS lookups please. return self.client_address[0] def log_request(*args, **kw): if not self.quiet: return WSGIRequestHandler.log_request(*args, **kw) handler_cls = self.options.get('handler_class', FixedHandler) server_cls = self.options.get('server_class', WSGIServer) if ':' in self.host: # Fix wsgiref for IPv6 addresses. if getattr(server_cls, 'address_family') == socket.AF_INET: class server_cls(server_cls): address_family = socket.AF_INET6 srv = make_server(self.host, self.port, app, server_cls, handler_cls) srv.serve_forever()
def start(self): self.load_config() self.load_contracts() self._web3 = Web3( RPCProvider( self.config['rpc']['host'], self.config['rpc']['port'], ssl=self.config['rpc']['ssl'] ) ) self.node = Node(self._web3) self.load_networks() ipport = ('', 5000) logger.info('Starting server') app = ApiApp(self) http_server = WSGIServer(ipport, app, log=None) logger.info('Server is running on {}'.format(ipport)) http_server.serve_forever()
def wsgi( self, port=None, host=None, max_accept=settings.GEVENT_MAX_CONNECTIONS, min_delay=0.1, max_delay=1, stop_timeout=1): if isinstance(port, int): self.port = port if isinstance(host, basestring): self.host = host wsgi_params = dict( max_accept=max_accept, min_delay=min_delay, max_delay=max_delay, stop_timeout=stop_timeout, get_url=lambda s: self.get_url() ) StandaloneServer = type( 'StandaloneServer', (WSGIServer, ), wsgi_params) return StandaloneServer((self.host, self.port), self)
def dev(p, kael_amqp): """ Example usage: \b $ kael-web dev $ kael-web dev -p 5000 --kael_amqp 'amqp://user:****@localhost:5672/api' """ app.config['AMQP_URI'] = kael_amqp print '\n', 'AMQP_URI:', str(kael_amqp), '\n' if not kael_amqp: raise click.BadParameter('Use --kael_amqp to set AMQP_URI (AMQP_URI not set)') app.debug = True print app.url_map print ' * Running on 0.0.0.0:{} (Press CTRL+C to quit)'.format(p) server = WSGIServer(("0.0.0.0", p), app) server.serve_forever()
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(app): linkero.printWellcome() linkero.createDB() if linkero.config["SSL"]["activate"]: gevent_server = WSGIServer((linkero.config["host"]["ip"], int(os.environ.get('PORT', linkero.config["host"]["port"]))), app, spawn=Pool(linkero.config["gevent"]["spawn"]), log='default' if (linkero.config["gevent"]["accessLog"] == True) else None, keyfile=linkero.config["SSL"]["key"], certfile=linkero.config["SSL"]["certificate"]) else: gevent_server = WSGIServer((linkero.config["host"]["ip"], int(os.environ.get('PORT', linkero.config["host"]["port"]))), app, spawn=Pool(linkero.config["gevent"]["spawn"], log = 'default' if (linkero.config["gevent"]["accessLog"] == True) else None)) gevent_server.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 set_sail(self, host, port): """ ??leics?? ????: ??! """ print "\033[34m{Sail}\033[0m [leics] is running on [http://%s:%d]" % (host, port) WSGIServer((host, port), self).serve_forever()
def main(): http_server = WSGIServer(('localhost', 0), hello_world, log=None) servelet = gevent.spawn(serve, http_server) # Wait for server being bound to socket. while True: if http_server.address[1] != 0: break gevent.sleep(0.05) client = gipc.start_process( target=child_test_wsgi_scenario_client, args=(http_server.address, )) client.join() servelet.kill() servelet.join()
def test_wsgi_scenario(self): from gevent.wsgi import WSGIServer def serve(http_server): http_server.serve_forever() def hello_world(environ, start_response): # Generate response in child process. with pipe() as (reader, writer): start_response('200 OK', [('Content-Type', 'text/html')]) rg = start_process( target=complchild_test_wsgi_scenario_respgen, args=(writer, )) response = reader.get() rg.join() assert rg.exitcode == 0 return [response] http_server = WSGIServer(('localhost', 0), hello_world) servelet = gevent.spawn(serve, http_server) # Wait for server being bound to socket. while True: if http_server.address[1] != 0: break gevent.sleep(0.05) client = start_process( target=complchild_test_wsgi_scenario_client, args=(http_server.address, )) client.join() assert client.exitcode == 0 servelet.kill() servelet.get() # get() is join and re-raises Exception.
def run(self, handler): # pragma: no cover import flup.server.fcgi self.options.setdefault('bindAddress', (self.host, self.port)) flup.server.fcgi.WSGIServer(handler, **self.options).run()
def run(self, handler): from gevent import wsgi, pywsgi, local if not isinstance(threading.local(), local.local): msg = "Bottle requires gevent.monkey.patch_all() (before import)" raise RuntimeError(msg) if not self.options.pop('fast', None): wsgi = pywsgi self.options['log'] = None if self.quiet else 'default' address = (self.host, self.port) server = wsgi.WSGIServer(address, handler, **self.options) if 'BOTTLE_CHILD' in os.environ: import signal signal.signal(signal.SIGINT, lambda s, f: server.stop()) server.serve_forever()
def run(self, app): # pragma: no cover from wsgiref.simple_server import make_server from wsgiref.simple_server import WSGIRequestHandler, WSGIServer import socket class FixedHandler(WSGIRequestHandler): def address_string(self): # Prevent reverse DNS lookups please. return self.client_address[0] def log_request(*args, **kw): if not self.quiet: return WSGIRequestHandler.log_request(*args, **kw) handler_cls = self.options.get('handler_class', FixedHandler) server_cls = self.options.get('server_class', WSGIServer) if ':' in self.host: # Fix wsgiref for IPv6 addresses. if getattr(server_cls, 'address_family') == socket.AF_INET: class server_cls(server_cls): address_family = socket.AF_INET6 self.srv = make_server(self.host, self.port, app, server_cls, handler_cls) self.port = self.srv.server_port # update port actual port (0 means random) try: self.srv.serve_forever() except KeyboardInterrupt: self.srv.server_close() # Prevent ResourceWarning: unclosed socket raise
def start(self, host='127.0.0.1', port=5001): self.wsgiserver = WSGIServer((host, port), self.flask_app, log=log, error_log=log) self.wsgiserver.start()
def run(self, handler): from gevent import wsgi, pywsgi, local if not isinstance(_lctx, local.local): msg = "Bottle requires gevent.monkey.patch_all() (before import)" raise RuntimeError(msg) if not self.options.get('fast'): wsgi = pywsgi log = None if self.quiet else 'default' wsgi.WSGIServer((self.host, self.port), handler, log=log).serve_forever()
def gevent_run(app, port=5000, log=None, error_log=None, address='', monkey_patch=True, start=True, **kwargs): # pragma: no cover """Run your app in gevent.wsgi.WSGIServer :param app: wsgi application, ex. Microservice instance :param port: int, listen port, default 5000 :param address: str, listen address, default: "" :param log: logger instance, default app.logger :param error_log: logger instance, default app.logger :param monkey_patch: boolean, use gevent.monkey.patch_all() for patching standard modules, default: True :param start: boolean, if True, server will be start (server.serve_forever()) :param kwargs: other params for WSGIServer(**kwargs) :return: server """ if log is None: log = app.logger if error_log is None: error_log = app.logger if monkey_patch: from gevent import monkey monkey.patch_all() from gevent.wsgi import WSGIServer http_server = WSGIServer((address, port), app, log=log, error_log=error_log, **kwargs) if start: http_server.serve_forever() return http_server
def run_server(): http_server = WSGIServer(('', 5000), DebuggedApplication(application)) http_server.serve_forever()
def run_api_server(app, port=5000): """ Runs a gevent server on a given port. Default port is 5000. :param app: WSGI-compliant app to be run :param port: port for the gevent server :type port: int :return: None """ log = logging.getLogger('api') http_server = WSGIServer(('', port), app, log=log) http_server.serve_forever()
def main(): parser = argparse.ArgumentParser() parser.add_argument('-p', '--port', help='server port', type=int, default=9000) args = parser.parse_args() http_server = WSGIServer(('', args.port), app) http_server.serve_forever()
def _start_flask(self): #self.app.run(self.ip, self.port, debug=False, use_reloader=False) #this should be a more production-fit http-server #self.app.logger.setLevel(logging.ERROR) self.http_server = WSGIServer((self.ip, self.port), self.app, log=open("/dev/null", "w") # This disables HTTP request logs to not mess up the CLI when e.g. the auto-updated dashboard is used ) self.http_server.serve_forever()
def main(): print 'Starting TRex HTTP proxy on port: ', server_config['port'] http_server = WSGIServer(('', server_config['port']), app) http_server.serve_forever() # Start web server
def run(self, handler): # pragma: no cover import flup.server.fcgi flup.server.fcgi.WSGIServer(handler, bindAddress=(self.host, self.port)).run()
def run(self, handler): from gevent import wsgi #from gevent.hub import getcurrent #self.set_context_ident(getcurrent, weakref=True) # see contextlocal wsgi.WSGIServer((self.host, self.port), handler).serve_forever()
