Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
docstring = str(
bytes(request_data["docstring"], "utf-8").decode("unicode_escape")
)
else:
docstring = None
endpoint_type = request_data["type"] if "type" in request_data else None
methods = request_data["methods"] if "methods" in request_data else []
dependencies = (
request_data["dependencies"] if "dependencies" in request_data else None
)
target = request_data["target"] if "target" in request_data else None
schema = request_data["schema"] if "schema" in request_data else None
src_path = request_data["src_path"] if "src_path" in request_data else None
target_path = get_query_object_path(
self.settings[SettingsParameters.StateFilePath], name, version
)
self.logger.log(logging.DEBUG, f"Checking source path {src_path}...")
_path_checker = _compile(r"^[\\\:a-zA-Z0-9-_~\s/\.\(\)]+$")
# copy from staging
if src_path:
if not isinstance(request_data["src_path"], str):
raise gen.Return("src_path must be a string.")
if not _path_checker.match(src_path):
raise gen.Return(
"Endpoint source path name can only contain: "
"a-z, A-Z, 0-9, underscore, hyphens and spaces."
)
yield self._copy_po_future(src_path, target_path)
elif endpoint_type != "alias":
raise gen.Return("src_path is required to add/update an " "endpoint.")
set_parameter(
SettingsParameters.CertificateFile, ConfigParameters.TABPY_CERTIFICATE_FILE
)
set_parameter(SettingsParameters.KeyFile, ConfigParameters.TABPY_KEY_FILE)
self._validate_transfer_protocol_settings()
# if state.ini does not exist try and create it - remove
# last dependence on batch/shell script
set_parameter(
SettingsParameters.StateFilePath,
ConfigParameters.TABPY_STATE_PATH,
default_val=os.path.join(pkg_path, "tabpy_server"),
)
self.settings[SettingsParameters.StateFilePath] = os.path.realpath(
os.path.normpath(
os.path.expanduser(self.settings[SettingsParameters.StateFilePath])
)
)
state_file_dir = self.settings[SettingsParameters.StateFilePath]
state_file_path = os.path.join(state_file_dir, "state.ini")
if not os.path.isfile(state_file_path):
state_file_template_path = os.path.join(
pkg_path, "tabpy_server", "state.ini.template"
)
logger.debug(
f"File {state_file_path} not found, creating from "
f"template {state_file_template_path}..."
)
shutil.copy(state_file_template_path, state_file_path)
logger.info(f"Loading state from state file {state_file_path}")
tabpy_state = _get_state_from_file(state_file_dir)
def get(self):
# do not check for authentication - this method
# is the only way for client to collect info about
# supported API versions and required features
self._add_CORS_header()
info = {}
info["description"] = self.tabpy_state.get_description()
info["creation_time"] = self.tabpy_state.creation_time
info["state_path"] = self.settings[SettingsParameters.StateFilePath]
info["server_version"] = self.settings[SettingsParameters.ServerVersion]
info["name"] = self.tabpy_state.name
info["versions"] = self.settings[SettingsParameters.ApiVersions]
self.write(json.dumps(info))
def init_ps_server(settings, tabpy_state):
logger.info("Initializing TabPy Server...")
existing_pos = tabpy_state.get_endpoints()
for (object_name, obj_info) in existing_pos.items():
try:
object_version = obj_info["version"]
get_query_object_path(
settings[SettingsParameters.StateFilePath], object_name, object_version
)
except Exception as e:
logger.error(
f"Exception encounted when downloading object: {object_name}"
f", error: {e}"
def get(self):
if self.should_fail_with_not_authorized():
self.fail_with_not_authorized()
return
path = self.settings[SettingsParameters.StateFilePath]
path = os.path.join(path, _QUERY_OBJECT_STAGING_FOLDER)
self.write({"path": path})
default_val="http",
)
self.settings[SettingsParameters.TransferProtocol] = self.settings[
SettingsParameters.TransferProtocol
].lower()
set_parameter(
SettingsParameters.CertificateFile, ConfigParameters.TABPY_CERTIFICATE_FILE
)
set_parameter(SettingsParameters.KeyFile, ConfigParameters.TABPY_KEY_FILE)
self._validate_transfer_protocol_settings()
# if state.ini does not exist try and create it - remove
# last dependence on batch/shell script
set_parameter(
SettingsParameters.StateFilePath,
ConfigParameters.TABPY_STATE_PATH,
default_val=os.path.join(pkg_path, "tabpy_server"),
)
self.settings[SettingsParameters.StateFilePath] = os.path.realpath(
os.path.normpath(
os.path.expanduser(self.settings[SettingsParameters.StateFilePath])
)
)
state_file_dir = self.settings[SettingsParameters.StateFilePath]
state_file_path = os.path.join(state_file_dir, "state.ini")
if not os.path.isfile(state_file_path):
state_file_template_path = os.path.join(
pkg_path, "tabpy_server", "state.ini.template"
)
logger.debug(
f"File {state_file_path} not found, creating from "
def write_state_config(state, settings, logger=logging.getLogger(__name__)):
if SettingsParameters.StateFilePath in settings:
state_path = settings[SettingsParameters.StateFilePath]
else:
msg = f"{ConfigParameters.TABPY_STATE_PATH} is not set"
logger.log(logging.CRITICAL, msg)
raise ValueError(msg)
logger.log(logging.DEBUG, f"State path is {state_path}")
state_key = os.path.join(state_path, "state.ini")
tmp_state_file = state_key
with open(tmp_state_file, "w") as f:
state.write(f)