# Copyright (c) 2022, 2025, Oracle and/or its affiliates. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License, version 2.0, # as published by the Free Software Foundation. # # This program is designed to work with certain software (including # but not limited to OpenSSL) that is licensed under separate terms, as # designated in a particular file or component or in included license # documentation. The authors of MySQL hereby grant you an additional # permission to link the program and your derivative works with the # separately licensed software that they have either included with # the program or referenced in the documentation. # # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See # the GNU General Public License, version 2.0, for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA import mrs_plugin.lib as lib import json import os from pathlib import Path def service_query_selection(**kwargs): service = kwargs.get("service") service_id = kwargs.get("service_id") url_host_name = kwargs.get("url_host_name") url_context_root = kwargs.get("url_context_root") if service_id is not None: return service_id if service is not None: return service if url_context_root is not None: return f"{url_context_root}" return None def schema_query_selection(**kwargs): schema = kwargs.get("schema") schema_id = kwargs.get("schema_id") if schema_id is not None: return schema_id if schema is not None: return schema return None def auth_app_query_selection(**kwargs): auth_app = kwargs.get("auth_app") auth_app_id = kwargs.get("auth_app_id") if auth_app_id is not None: return auth_app_id if auth_app is not None: return auth_app return None def resolve_service(session, service_query:str | bytes=None, required:bool=True, auto_select_single:bool=False): service = None if service_query: if isinstance(service_query, bytes): # Check if given service exists by searching its id service = lib.services.get_service( service_id=service_query, session=session) else: # Check if the service exists by host and context root url_host_name, url_context_root = service_query.split("/", 1) url_context_root = f"/{url_context_root}" service = lib.services.get_service( url_host_name=url_host_name, url_context_root=url_context_root, session=session) if not service: service = lib.services.get_current_service(session) services = lib.services.get_services(session) if len(services) == 1 and auto_select_single: # If there only is one service and auto_select_single is True, take the single service service = services[0] if not service and lib.core.get_interactive_default(): print("MRS - Service Listing\n") service = lib.core.prompt_for_list_item( item_list=services, prompt_caption=("Please select a service index or type " "'hostname/root_context'"), item_name_property="host_ctx", given_value=None, print_list=True) if not service and required: raise Exception("Operation cancelled. Unable to identify target service.") return service def resolve_schema(session, schema_query:str | bytes=None, service_query:str | bytes=None, required:bool=True): schema = None service = None if schema_query is not None: if isinstance(schema_query, bytes): schema = lib.schemas.get_schema(session=session, schema_id=schema_query) elif isinstance(schema_query, str): url_host_name, url_context_root, request_path = schema_query.split("/") service = lib.services.get_service(session, url_host_name=url_host_name, url_context_root=f"/{url_context_root}") if service is not None: schema = lib.schemas.get_schema(session, service_id=service["id"], request_path=f"/{request_path}") if schema: return schema if lib.core.get_interactive_default(): service = resolve_service(session, service_query) schemas = lib.schemas.get_schemas(session, service["id"]) schema = lib.core.prompt_for_list_item( item_list=schemas, prompt_caption='Please enter the name or index of a schema: ', item_name_property="name", print_list=True) if not schema and required: raise Exception("Cancelling operation. Could not determine the schema.") return schema def resolve_db_object(session, db_object_id:bytes=None, schema_id:bytes=None, service_id:bytes=None, required:bool=True): db_object = None if db_object_id: db_object = lib.db_objects.get_db_object(session, db_object_id) if db_object: return db_object if lib.core.get_interactive_default(): schema = resolve_schema(session, schema_query=schema_id, service_query=service_id) db_objects = lib.db_objects.get_db_objects(session, schema["id"]) db_object = lib.core.prompt_for_list_item( item_list=db_objects, prompt_caption='Please enter the name or index of a db_object: ', item_name_property="name", print_list=True) if not db_object and required: raise Exception("Cancelling operation. Could not determine the db_object.") return db_object def resolve_content_set(session, content_set_id:bytes=None, service_id:bytes=None, required:bool=True): content_set = None if content_set_id: content_set = lib.content_sets.get_content_set(session, content_set_id=content_set_id) if content_set: return content_set if lib.core.get_interactive_default(): if not service_id: service = resolve_service(session, service_id) content_sets = lib.content_sets.get_content_sets(session, service["id"]) content_set = lib.core.prompt_for_list_item( item_list=content_sets, prompt_caption='Please enter the name or index of a content set: ', item_name_property="request_path", print_list=True) if not content_set and required: raise Exception("Cancelling operation. Could not determine the content set.") return content_set def resolve_content_file(session, content_file_id:bytes=None, content_set_id:bytes=None, service_id:bytes=None, required:bool=True): content_file = None if content_file_id: content_file = lib.content_files.get_content_file(session, content_file_id) if content_file: return content_file if lib.core.get_interactive_default(): content_set = resolve_content_set(session, content_set_id, service_id) content_files = lib.content_files.get_content_files(session, service_id=content_set["id"]) content_file = lib.core.prompt_for_list_item( item_list=content_files, prompt_caption='Please enter the name or index of a content file: ', item_name_property="request_path", print_list=True) if not content_file and required: raise Exception("Cancelling operation. Could not determine the content file.") return content_file def user_query_selection(**kwargs): user_id = kwargs.get("user_id") user = kwargs.get("user") if user_id is not None: return user_id if user is not None: return user return None def resolve_user(session, user_query:str | bytes): if isinstance(user_query, bytes): return lib.users.get_user(session, user_id=user_query) if isinstance(user_query, str): url_host_name, url_context_root, auth_app_name, user_name = user_query.split("/") service = lib.services.get_service(session, url_host_name=url_host_name, url_context_root=f"/{url_context_root}") if not service: raise Exception("service not found") auth_app = lib.auth_apps.get_auth_app(session, name=auth_app_name) if not auth_app: raise Exception("auth_app not found") user = lib.users.get_user(session, service_id=service["id"], auth_app_id=auth_app["id"], user_name=user_name) if not user: raise Exception("user not found") return user return None def role_query_selection(**kwargs): role_id = kwargs.get("role_id") role_name = kwargs.get("role_id") role_query = kwargs.get("role") if role_id is not None: return role_id if isinstance(role_query, bytes) or isinstance(role_query, str): return role_query if role_name is not None: return role_name return None def resolve_role(session, role_query:str | bytes): if isinstance(role_query, bytes): return lib.roles.get_role(session, role_id=role_query) role_query_list = role_query.split("/") service_id = None role_name = None if len(role_query_list) == 3: url_host_name = role_query_list[0] url_context_root = role_query_list[1] role_name = role_query_list[2] service = lib.services.get_service(session, url_host_name=url_host_name, url_context_root=url_context_root) service_id = service["id"] elif len(role_query_list) == 1: role_name = role_query_list[0] roles = lib.roles.get_roles(session, service_id) for role in roles: if role["caption"] == role_name: return role return None def resolve_options(options, default = None): # it should be possible to override the default value with an empty dict if options is not None: return options if lib.core.get_interactive_default(): entries = [ "Default Service Options for Development", "No options", "Custom options" ] selection = lib.core.prompt_for_list_item( item_list=entries, prompt_caption=("Please select how to initialize the options [Default Service Options for Development]"), prompt_default_value=entries[0], given_value=None, print_list=True) if selection == "Default Service Options for Development": return default elif selection == "No options": return None elif selection == "Custom options": return json.loads(lib.core.prompt("Options (in JSON format): ")) return None def resolve_auth_app(session, auth_app_query:str | bytes=None, service_query:str | bytes=None, required:bool=True): if isinstance(auth_app_query, bytes): return lib.auth_apps.get_auth_app(session, auth_app_query) service = resolve_service(session, service_query=service_query) auth_app_list = lib.auth_apps.get_auth_apps(session, service["id"]) if len(auth_app_list) == 0: raise Exception("No Authentication Apps found for this service.") if len(auth_app_list) == 1: return auth_app_list[0] selection = lib.core.prompt_for_list_item( item_list=auth_app_list, prompt_caption='Please enter the name or index of an auth_app: ', item_name_property="name", print_list=True) if not selection and required: raise Exception("Cancelling operation. Could not determine the auth_app.") return selection def resolve_file_path(file_path:str=None, required:bool=True): if file_path is None and lib.core.get_interactive_default(): file_path = lib.core.prompt("Please set the file path", { "defaultValue": None, }) if not file_path and required: raise Exception("Cancelling operation. Could not determine the file path.") if file_path.startswith("~"): file_path = os.path.expanduser(file_path) if not os.path.isabs(file_path): return Path(Path.home() / file_path).as_posix() return Path(file_path).as_posix() def resolve_overwrite_file(file_path:str, overwrite:bool) -> None: if not os.path.exists(file_path): return if os.path.isdir(file_path): raise Exception("Cancelling operation. Path already exists and it's a directory.") if overwrite is None and lib.core.get_interactive_default(): overwrite = lib.core.prompt(f"Overwrite {file_path}? [y/N]: ", {'defaultValue': 'n'}).strip().lower() == "y" if overwrite is False: raise Exception(f"Cancelling operation. File '{file_path}' already exists.")