Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _makeOne(
self,
application=dummy_app,
host="127.0.0.1",
port=0,
_dispatcher=None,
adj=None,
map=None,
_start=True,
_sock=None,
_server=None,
):
from waitress.server import create_server
self.inst = create_server(
application,
host=host,
port=port,
map=map,
_dispatcher=_dispatcher,
_start=_start,
_sock=_sock,
)
return self.inst
def _makeOneWithMulti(
self, adj=None, _start=True, app=dummy_app, listen="127.0.0.1:0 127.0.0.1:0"
):
sock = DummySock()
task_dispatcher = DummyTaskDispatcher()
map = {}
from waitress.server import create_server
self.inst = create_server(
app,
listen=listen,
map=map,
_dispatcher=task_dispatcher,
_start=_start,
_sock=sock,
)
return self.inst
def _makeOne(self, _start=True, _sock=None):
from waitress.server import create_server
self.inst = create_server(
dummy_app,
map={},
_start=_start,
_sock=_sock,
_dispatcher=DummyTaskDispatcher(),
unix_socket=self.unix_socket,
unix_socket_perms="600",
)
return self.inst
def start():
"""Start webserver"""
logged_app = TransLogger(wsgi.application)
# Work with systemd socket activation
if "LISTEN_FDNAMES" in os.environ and not config.LISTEN:
wsgi_server = NonBindingServer(
logged_app, _sock=socket.fromfd(3, socket.AF_UNIX, socket.SOCK_STREAM)
)
elif config.LISTEN.startswith("/"):
wsgi_server = waitress.server.create_server(
logged_app, unix_socket=config.LISTEN
)
elif config.LISTEN:
wsgi_server = waitress.server.create_server(logged_app, listen=config.LISTEN)
else:
raise RuntimeError("No socket to listen on. `LISTEN` config value is empty.")
if wsgi.warmup.status_code != 200:
log.error(
"Warmup failed with status code %s. Shutting down.", wsgi.warmup.status_code
)
exit(1)
wsgi_server.print_listen("Serving on {}:{}")
systemd_notify("READY=1")
wsgi_server.run()
def serve(app, **kw):
_server = kw.pop("_server", create_server) # test shim
_quiet = kw.pop("_quiet", False) # test shim
_profile = kw.pop("_profile", False) # test shim
if not _quiet: # pragma: no cover
# idempotent if logging has already been set up
logging.basicConfig()
server = _server(app, **kw)
if not _quiet: # pragma: no cover
server.print_listen("Serving on http://{}:{}")
if _profile: # pragma: no cover
profile("server.run()", globals(), locals(), (), False)
else:
server.run()
def createServer(application, logger, **kw):
global server
server = create_server(
TransLogger(application, logger=logger),
trusted_proxy='*',
trusted_proxy_headers=('x-forwarded-for',),
clear_untrusted_proxy_headers=True,
**kw
)
if not hasattr(server, 'addr'):
server.addr = server.adj.listen[0][3]
elif not server.addr:
server.addr = server.sockinfo[3]
return server
def serve(app, **kw):
_server = kw.pop('_server', create_server) # test shim
_quiet = kw.pop('_quiet', False) # test shim
_profile = kw.pop('_profile', False) # test shim
if not _quiet: # pragma: no cover
# idempotent if logging has already been set up
logging.basicConfig()
server = _server(app, **kw)
if not _quiet: # pragma: no cover
server.print_listen('Serving on http://{}:{}')
if _profile: # pragma: no cover
profile('server.run()', globals(), locals(), (), False)
else:
server.run()
def custom_waitress_create_server(
self,
application,
map=None,
_start=True, # test shim
_sock=None, # test shim
_dispatcher=None, # test shim
**kw):
import waitress.server
self._waitress_server = waitress.server.create_server(
application, map=map, _start=_start, _sock=_sock,
_dispatcher=_dispatcher, **kw)
return self._waitress_server