我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用twisted.web.wsgi.server()。
def _cli_parse(args): # pragma: no coverage from argparse import ArgumentParser parser = ArgumentParser(usage="usage: %sprog [options] package.module:app") opt = parser.add_argument opt('app', help='WSGI app entry point.') opt("--version", action="store_true", help="show version number.") opt("-b", "--bind", metavar="ADDRESS", help="bind socket to ADDRESS.") opt("-s", "--server", default='wsgiref', help="use SERVER as backend.") opt("-p", "--plugin", action="append", help="install additional plugin/s.") opt("-c", "--conf", action="append", metavar="FILE", help="load config values from FILE.") opt("-C", "--param", action="append", metavar="NAME=VALUE", help="override config values.") opt("--debug", action="store_true", help="start server in debug mode.") opt("--reload", action="store_true", help="auto-reload on file changes.") cli_args = parser.parse_args(args) return cli_args, parser
def urlparts(self): """ The :attr:`url` string as an :class:`urlparse.SplitResult` tuple. The tuple contains (scheme, host, path, query_string and fragment), but the fragment is always empty because it is not visible to the server. """ env = self.environ http = env.get('HTTP_X_FORWARDED_PROTO') \ or env.get('wsgi.url_scheme', 'http') host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST') if not host: # HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients. host = env.get('SERVER_NAME', '127.0.0.1') port = env.get('SERVER_PORT') if port and port != ('80' if http == 'http' else '443'): host += ':' + port path = urlquote(self.fullpath) return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def run(self, handler): # pragma: no cover from cherrypy import wsgiserver self.options['bind_addr'] = (self.host, self.port) self.options['wsgi_app'] = handler certfile = self.options.get('certfile') if certfile: del self.options['certfile'] keyfile = self.options.get('keyfile') if keyfile: del self.options['keyfile'] server = wsgiserver.CherryPyWSGIServer(**self.options) if certfile: server.ssl_certificate = certfile if keyfile: server.ssl_private_key = keyfile try: server.start() finally: server.stop()
def urlparts(self): ''' The :attr:`url` string as an :class:`urlparse.SplitResult` tuple. The tuple contains (scheme, host, path, query_string and fragment), but the fragment is always empty because it is not visible to the server. ''' env = self.environ http = env.get('HTTP_X_FORWARDED_PROTO') or env.get('wsgi.url_scheme', 'http') host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST') if not host: # HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients. host = env.get('SERVER_NAME', '127.0.0.1') port = env.get('SERVER_PORT') if port and port != ('80' if http == 'http' else '443'): host += ':' + port path = urlquote(self.fullpath) return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def urlparts(self): """ The :attr:`url` string as an :class:`urlparse.SplitResult` tuple. The tuple contains (scheme, host, path, query_string and fragment), but the fragment is always empty because it is not visible to the server. """ env = self.environ http = env.get('HTTP_X_FORWARDED_PROTO') or env.get('wsgi.url_scheme', 'http') host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST') if not host: # HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients. host = env.get('SERVER_NAME', '127.0.0.1') port = env.get('SERVER_PORT') if port and port != ('80' if http == 'http' else '443'): host += ':' + port path = urlquote(self.fullpath) return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def run(self, handler): from eventlet import wsgi, listen, patcher if not patcher.is_monkey_patched(os): msg = "Bottle requires eventlet.monkey_patch() (before import)" raise RuntimeError(msg) socket_args = {} for arg in ('backlog', 'family'): try: socket_args[arg] = self.options.pop(arg) except KeyError: pass address = (self.host, self.port) try: wsgi.server(listen(address, **socket_args), handler, log_output=(not self.quiet)) except TypeError: # Fallback, if we have old version of eventlet wsgi.server(listen(address), handler)
def _cli_parse(args): from optparse import OptionParser parser = OptionParser( usage="usage: %prog [options] package.module:app") opt = parser.add_option opt("--version", action="store_true", help="show version number.") opt("-b", "--bind", metavar="ADDRESS", help="bind socket to ADDRESS.") opt("-s", "--server", default='wsgiref', help="use SERVER as backend.") opt("-p", "--plugin", action="append", help="install additional plugin/s.") opt("-c", "--conf", action="append", metavar="FILE", help="load config values from FILE.") opt("-C", "--param", action="append", metavar="NAME=VALUE", help="override config values.") opt("--debug", action="store_true", help="start server in debug mode.") opt("--reload", action="store_true", help="auto-reload on file changes.") opts, args = parser.parse_args(args[1:]) return opts, args, parser
def _reloader_child(server, app, interval): ''' Start the server and check for modified files in a background thread. As soon as an update is detected, KeyboardInterrupt is thrown in the main thread to exit the server loop. The process exists with status code 3 to request a reload by the observer process. If the lockfile is not modified in 2*interval second or missing, we assume that the observer process died and exit with status code 1 or 2. ''' lockfile = os.environ.get('BOTTLE_LOCKFILE') bgcheck = FileCheckerThread(lockfile, interval) try: bgcheck.start() server.run(app) except KeyboardInterrupt: pass bgcheck.status, status = 5, bgcheck.status bgcheck.join() # bgcheck.status == 5 --> silent exit if status: sys.exit(status)
def urlparts(self): ''' The :attr:`url` string as an :class:`urlparse.SplitResult` tuple. The tuple contains (scheme, host, path, query_string and fragment), but the fragment is always empty because it is not visible to the server. ''' env = self.environ http = env.get('wsgi.url_scheme', 'http') host = env.get('HTTP_X_FORWARDED_HOST') or env.get('HTTP_HOST') if not host: # HTTP 1.1 requires a Host-header. This is for HTTP/1.0 clients. host = env.get('SERVER_NAME', '127.0.0.1') port = env.get('SERVER_PORT') if port and port != ('80' if http == 'http' else '443'): host += ':' + port path = urlquote(self.fullpath) return UrlSplitResult(http, host, path, env.get('QUERY_STRING'), '')
def _cli_patch(cli_args): # pragma: no coverage parsed_args, _ = _cli_parse(cli_args) opts = parsed_args if opts.server: if opts.server.startswith('gevent'): import gevent.monkey gevent.monkey.patch_all() elif opts.server.startswith('eventlet'): import eventlet eventlet.monkey_patch()
def script_name(self): """ The initial portion of the URL's `path` that was removed by a higher level (server or routing middleware) before the application was called. This script path is returned with leading and tailing slashes. """ script_name = self.environ.get('SCRIPT_NAME', '').strip('/') return '/' + script_name + '/' if script_name else '/'
def auth(self): """ HTTP authentication data as a (user, password) tuple. This implementation currently supports basic (not digest) authentication only. If the authentication happened at a higher level (e.g. in the front web-server or a middleware), the password field is None, but the user field is looked up from the ``REMOTE_USER`` environ variable. On any errors, None is returned. """ basic = parse_auth(self.environ.get('HTTP_AUTHORIZATION', '')) if basic: return basic ruser = self.environ.get('REMOTE_USER') if ruser: return (ruser, None) return None
def run(self, handler): from meinheld import server server.listen((self.host, self.port)) server.run(handler)
def run(self, handler): # pragma: no cover import tornado.wsgi, tornado.httpserver, tornado.ioloop container = tornado.wsgi.WSGIContainer(handler) server = tornado.httpserver.HTTPServer(container) server.listen(port=self.port, address=self.host) tornado.ioloop.IOLoop.instance().start()
def run(self, handler): from twisted.web import server, wsgi from twisted.python.threadpool import ThreadPool from twisted.internet import reactor thread_pool = ThreadPool() thread_pool.start() reactor.addSystemEventTrigger('after', 'shutdown', thread_pool.stop) factory = server.Site(wsgi.WSGIResource(reactor, thread_pool, handler)) reactor.listenTCP(self.port, factory, interface=self.host) if not reactor.running: reactor.run()
def run(self, handler): from gevent import pywsgi, local if not isinstance(threading.local(), local.local): msg = "Bottle requires gevent.monkey.patch_all() (before import)" raise RuntimeError(msg) if self.quiet: self.options['log'] = None address = (self.host, self.port) server = pywsgi.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, handler): from rocket import Rocket server = Rocket((self.host, self.port), 'wsgi', {'wsgi_app': handler}) server.start()
def script_name(self): ''' The initial portion of the URL's `path` that was removed by a higher level (server or routing middleware) before the application was called. This script path is returned with leading and tailing slashes. ''' script_name = self.environ.get('SCRIPT_NAME', '').strip('/') return '/' + script_name + '/' if script_name else '/'
