Python wsgiref.simple_server 模块，make_server() 实例源码

def serve(store):
    args = store.args
    abe = Aml(store, args)

    if args.host or args.port:
        # HTTP server.
        if args.host is None:
            args.host = "localhost"
        from wsgiref.simple_server import make_server
        port = int(args.port or 80)
        httpd = make_server(args.host, port, abe )
        print "Listening on http://" + args.host + ":" + str(port)
        try:
            httpd.serve_forever()
        except:
            httpd.shutdown()
            raise

def test_make_client(self):
        test = self.Test2(value=10)
        app = bottle_api.make_app(('/api', test))
        server = make_server('localhost', 10001, app)
        process = Thread(
            target=server.serve_forever)
        process.start()

        time.sleep(1)  # Wait for the app to run

        c = client.make_client(
            instance=test, base_url='http://localhost:10001/api')
        data = c.endpoint(arg1='test')
        expected_data = {'works': True, 'arg1': 'test', 'value': 10}
        self.assertEqual(data, expected_data)

        with self.assertRaises(AttributeError):
            c.invalid_endpoint()

        test.endpoint(arg1='test')  # For 100% coverage :D

        app.close()
        server.shutdown()
        process.join()

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 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 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 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 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 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 run(self, handler):
        self.certfile = 'keys/server.pem'

        from wsgiref.simple_server import make_server, WSGIRequestHandler
        import ssl
        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)
        try:
            with open(self.certfile):
                pass
            srv.socket = ssl.wrap_socket(
                srv.socket,
                certfile=self.certfile
                )
            srv.serve_forever()
        except IOError as e:
            logging.error("Unable to open Certificate file at location {} {}".
                  format(self.certfile, e))
            raise

def __init__(self, host="127.0.0.1", port=8545, *args, **kwargs):
        if not is_testrpc_available():
            raise Exception("`TestRPCProvider` requires the `eth-testrpc` package to be installed")
        from testrpc.server import get_application

        application = get_application()

        self.server = make_server(
            host,
            port,
            application,
        )

        self.thread = spawn(self.server.serve_forever)
        endpoint_uri = 'http://{0}:{1}'.format(host, port)

        super(TestRPCProvider, self).__init__(endpoint_uri, *args, **kwargs)

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 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 run_client(queue):
    try:
        app = ClientApplication(
            callback_url="http://127.0.0.1:15487/callback",
            client_id="abc",
            client_secret="xyz",
            provider_url="http://127.0.0.1:15486")

        httpd = make_server('', 15487, app,
                            handler_class=NoLoggingHandler)

        queue.put({"result": 0})

        httpd.serve_forever()
    except Exception as e:
        queue.put({"result": 1, "error_message": str(e)})

def run(self):
        """
        Open WSGI server to listen to HOST_BASE address
        """
        print("Running")
        parts = urlparse(self.hostbase)
        domain, port = parts.netloc.split(":")
        print(domain, port)
        self.srv = make_server(domain, int(port), self.app)
        try:
            self.srv.serve_forever()
        except:
            # We are a background thread so we have problems to interrupt tests in the case of error
            import traceback
            traceback.print_exc()
            # Failed to start
            self.srv = None

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 run(self, app):
        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

        self.srv = make_server(self.host, self.port, app, server_cls, handler_cls)
        self.srv.serve_forever()
        while not self.get_port(): time.sleep(0.001)

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 __init__(self, config, port):
        super(ReahlWebServer, self).__init__()
        self.in_separate_thread = None
        self.running = False
        self.handlers = {}
        self.httpd_thread = None
        certfile = pkg_resources.resource_filename(__name__, 'reahl_development_cert.pem')
        self.reahl_wsgi_app = WrappedApp(ReahlWSGIApplication(config))
        try:
            https_port = port+363
            self.httpd = ReahlWSGIServer.make_server('', port, self.reahl_wsgi_app)
            self.httpsd = SSLCapableWSGIServer.make_server('', https_port, certfile, self.reahl_wsgi_app)
        except socket.error as ex:
            message = ('Caught socket.error: %s\nThis means that another process is using one of these ports: %s, %s. ' % (ex, port, https_port)) \
                     +'\nIf this happens while running tests, it probably means that a browser client did not close its side of a connection to a previous server you had running - and that the server socket now sits in TIME_WAIT state. Is there perhaps a browser hanging around from a previous run? I have no idea how to fix this automatically... see http://hea-www.harvard.edu/~fine/Tech/addrinuse.html'

            raise AssertionError(message)

def runServer(kodi = '', kodi_home = '', startBrowser=False):
    importer = ksi.KodiScriptImporter(kodi, kodi_home)
    importer.install(True)
    kodiSrv = KodiServer(importer)
    wsgiApp = partial(application, server=kodiSrv)

    server_address = ('localhost', 5000)
    httpd = make_server(
        server_address[0],  # The host name
        server_address[1],  # A port number where to wait for the request
        wsgiApp  # The application object name, in this case a function
    )
    srvThread = threading.Thread(target=httpd.serve_forever)
    if startBrowser:
        webbrowser.open('http://{}:{}'.format(*server_address))
    srvThread.start()
    return httpd

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 run(self, handler):
        from wsgiref.simple_server import make_server, WSGIRequestHandler
        import ssl
        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.socket = ssl.wrap_socket(
            srv.socket,
            keyfile=self.keyfile,
            certfile=self.certfile,
            ca_certs=self.cafile,
            cert_reqs=ssl.CERT_REQUIRED,
            server_side=True)
        srv.serve_forever()

def run(self, handler):
        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
        try:
            self.server = make_server(self.host, self.port, handler,
                                      **self.options)
            self.romana_http.wsgi_server_started = True
            logging.info("HTTP server: Started to listen...")
            self.server.serve_forever()
        except socket.error as e:
            logging.fatal("HTTP server: Cannot open socket "
                          "(error %d: %s)... " %
                          (e.errno, e.strerror))

def run(self, handler):
        from wsgiref.simple_server import make_server, WSGIRequestHandler
        import ssl
        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.socket = ssl.wrap_socket (
         srv.socket,
         keyfile='mfa_slipstream.pem',
         certfile='mfa_slipstream.pem',  # path to certificate
         server_side=True)
        srv.serve_forever()


##########################################################################
# UTILITY FUNCTIONS
##########################################################################

def run_dev_server():  # pragma: no cover
    """Usage: python -m nuka.report <type> <data_filename>"""
    from wsgiref.simple_server import make_server
    engine = nuka.config.get_template_engine()
    type, filename = sys.argv[-2:]

    def app(environ, start_response):
        start_response('200 OK',
                       [('Content-Type', 'text/html; charset=utf8')])
        template = engine.get_template('reports/{0}.html.j2'.format(type))
        with open(filename) as fd:
            data = json.load(fd)
        html = template.render(dict(data=data, dumped_data=json.dumps(data)))
        return [html.encode('utf8')]

    httpd = make_server('', 8000, app)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass

def http(tasks, bind):
    """Http interface using built-in simple wsgi server"""
    from wsgiref.simple_server import make_server
    from .utils import load_manager
    from .http import Application

    host, _, port = bind.partition(':')
    app = Application(load_manager(tasks))
    httpd = make_server(host, int(port), app)
    print('Listen on {}:{} ...'.format(host or '0.0.0.0', port), file=sys.stderr)
    httpd.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, handler):
        class QuietHandler(WSGIRequestHandler):
            def log_request(*args, **kw):
                pass

            def log_error(self, format, *args):
                pass
        self.options['handler_class'] = QuietHandler
        srv = make_server(self.host, self.port, handler, **self.options)
        srv.serve_forever()

def run(self, handler):
        class QuietHandler(WSGIRequestHandler):
            def log_request(*args, **kw):
                pass

            def log_error(self, format, *args):
                pass
        self.options['handler_class'] = QuietHandler
        srv = make_server(self.host, self.port, handler, **self.options)
        srv.socket = ssl.wrap_socket(srv.socket, certfile=CERT_PATH, server_side=True)
        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 = FixedHandler
        server_cls = 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):
        """Run the Process"""
        self.runner = TaskRunner()
        self.runner_api = TaskRunnerAPI(process=self, runner=self.runner)

        self.app = bottle_api.make_app(('/api', self.runner_api))
        self.server = make_server(self.host, self.port, self.app)

        self.runner.start()
        self.server.serve_forever()
        self.runner.join()

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()

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