Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
)
if not key_is_set and default_val is not None:
self.settings[settings_key] = default_val
key_is_set = True
logger.debug(
f"Parameter {settings_key} set to "
f'"{self.settings[settings_key]}" '
"from default value"
)
if not key_is_set:
logger.debug(f"Parameter {settings_key} is not set")
set_parameter(
SettingsParameters.Port, ConfigParameters.TABPY_PORT, default_val=9004
)
set_parameter(SettingsParameters.ServerVersion, None, default_val=__version__)
set_parameter(
SettingsParameters.EvaluateTimeout,
ConfigParameters.TABPY_EVALUATE_TIMEOUT,
default_val=30,
)
try:
self.settings[SettingsParameters.EvaluateTimeout] = float(
self.settings[SettingsParameters.EvaluateTimeout]
)
except ValueError:
logger.warning(
"Evaluate timeout must be a float type. Defaulting "
def initialize(self, app):
self.tabpy_state = app.tabpy_state
# set content type to application/json
self.set_header("Content-Type", "application/json")
self.protocol = self.settings[SettingsParameters.TransferProtocol]
self.port = self.settings[SettingsParameters.Port]
self.python_service = app.python_service
self.credentials = app.credentials
self.username = None
self.password = None
self.eval_timeout = self.settings[SettingsParameters.EvaluateTimeout]
self.logger = ContextLoggerWrapper(self.request)
self.logger.enable_context_logging(
app.settings[SettingsParameters.LogRequestContext]
)
self.logger.log(logging.DEBUG, "Checking if need to handle authentication")
self.not_authorized = not self.handle_authentication("v1")
def initialize(self, app):
super(ManagementHandler, self).initialize(app)
self.port = self.settings[SettingsParameters.Port]
init_model_evaluator(self.settings, self.tabpy_state, self.python_service)
protocol = self.settings[SettingsParameters.TransferProtocol]
ssl_options = None
if protocol == "https":
ssl_options = {
"certfile": self.settings[SettingsParameters.CertificateFile],
"keyfile": self.settings[SettingsParameters.KeyFile],
}
elif protocol != "http":
msg = f"Unsupported transfer protocol {protocol}."
logger.critical(msg)
raise RuntimeError(msg)
application.listen(
self.settings[SettingsParameters.Port],
ssl_options=ssl_options,
max_buffer_size=max_request_size,
max_body_size=max_request_size,
)
logger.info(
"Web service listening on port "
f"{str(self.settings[SettingsParameters.Port])}"
)
tornado.ioloop.IOLoop.instance().start()