def main():
logging.register_options(CONF)
CONF(sys.argv[1:],
project='craton-api',
default_config_files=[])
logging.setup(CONF, 'craton-api')
app = api.load_app()
host, port = cfg.CONF.api.host, cfg.CONF.api.port
srv = simple_server.make_server(host, port, app)
LOG.info("Starting API server in PID: %s" % os.getpid())
srv.serve_forever()
python类make_server()的实例源码
def run_server(app, host, port):
click.echo('Start: {host}:{port}'.format(host=host, port=port))
httpd = make_server(host, port, app)
httpd.serve_forever()
# For reloading server when detected python files changes.
def listen(self):
"""Self-host using 'bind' and 'port' from the WSGI config group."""
msgtmpl = (u'Serving on host %(host)s:%(port)s')
host = CONF.wsgi.wsgi_host
port = CONF.wsgi.wsgi_port
LOG.info(msgtmpl,
{'host': host, 'port': port})
server_cls = self._get_server_cls(host)
httpd = simple_server.make_server(host,
port,
self.app,
server_cls)
httpd.serve_forever()
def main():
service.prepare_service(sys.argv)
cfg.CONF.log_opt_values(LOG, logging.DEBUG)
# Set source of model files
service_list = particle_generator.get_service_list()
LOG.info("Service List: %s" % service_list)
LOG.info("Generating DB Classes")
particle_generator.build_sql_models(service_list)
# API is generated during the setup_app phase.
LOG.info("Generating API Classes")
app = api_app.setup_app()
# Create the WSGI server and start it
host, port = cfg.CONF.api.host, cfg.CONF.api.port
srv = simple_server.make_server(host, port, app)
LOG.info('Starting server in PID %s' % os.getpid())
LOG.debug("Configuration:")
if host == '0.0.0.0':
LOG.info(('serving on 0.0.0.0:%(port)s, '
'view at http://127.0.0.1:%(port)s') %
dict(port=port))
else:
LOG.info('serving on http://%(host)s:%(port)s' %
dict(host=host, port=port))
start_sync_thread(etcd_host=cfg.CONF.api.etcd_host,
etcd_port=cfg.CONF.api.etcd_port)
srv.serve_forever()
def serve(address, root, **kwargs): # pragma: no cover
""" Serve static files from root directory. """
app = Rheostatic(root, **kwargs)
server = make_server(address[0], address[1], validator(app))
try:
print('Starting server at http://%s:%d/...' % address)
print('Serving files from %s' % app.root)
print('Press ctrl+c to stop.')
server.serve_forever()
except KeyboardInterrupt:
print('Quiting...')
def bootstrap(raml_file='api_schema.raml'):
logger.info('Creating development server')
app = build_wsgi_app(raml_file)
logger.info('Starting server on port 8080')
server = make_server('0.0.0.0', 8080, app)
server.serve_forever()
def run(host, port):
httpd = make_server(host, port, app)
print("Serving on port %s..." % (port))
httpd.serve_forever()
def main():
host = CONF.host
port = CONF.api_port
# Create the WSGI application object.
wsgi_application = wsgi_app.setup_app()
# Create the Simple process server.
server = simple_server.make_server(host, port, wsgi_application)
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 run(self, app):
self.server = make_server(self.host, self.port, app)
self.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 wsgiref(app, address, **options): # pragma: no cover
from wsgiref.simple_server import make_server, WSGIRequestHandler
options = {}
class QuietHandler(WSGIRequestHandler):
def log_request(*args, **kw):
pass
options['handler_class'] = QuietHandler
srv = make_server(address[0], address[1], app, **options)
srv.serve_forever()
def run(self, port=8000, host='', certfile=None, keyfile=None):
import ssl
from wsgiref.simple_server import make_server
httpd = make_server(host, port, self.get_wsgi_application())
if certfile and keyfile:
httpd.socket = ssl.wrap_socket(
httpd.socket,
certfile=certfile,
keyfile=keyfile,
server_side=True
)
print("Serving on port %s:%d..." % (host, port))
httpd.serve_forever()
def start(self):
server = simple_server.make_server('', 8000, app.api)
self.process = multiprocessing.Process(target=server.serve_forever,
name='test-server')
self.process.daemon = True
self.process.start()
self.process.join(1)
def loop():
# ????????IP????????8000??????application:
httpd = make_server('', 8000, application)
print("Serving HTTP on port 8000...")
# ????HTTP??:
httpd.serve_forever()
def runserver():
logging.basicConfig(level=LOGGING_LEVEL)
logger.setLevel(LOGGING_LEVEL)
try:
UnicamPI.initiate()
server = make_server(SERVER, PORT, UnicamPI.app)
logger.info('serving at http://%s:%i' % (SERVER, PORT))
server.serve_forever()
except KeyboardInterrupt:
pass
def build_web_server(app, port, host=None):
server = wsgiserver.make_server(
host or '', port, app,
server_class=WebServer,
handler_class=_RequestHandler)
return server
def run(self, handler): # pragma: no cover
from wsgiref.simple_server import make_server, WSGIRequestHandler
if self.quiet:
class QuietHandler(WSGIRequestHandler):
def log_request(*args, **kw): pass
self.options['handler_class'] = QuietHandler
srv = make_server(self.host, self.port, handler, **self.options)
srv.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 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
