import json import os import ast import sys from pathlib import Path import toml import configparser import subprocess import shlex from distutils.dir_util import remove_tree from shutil import copytree, ignore_patterns from unittest.mock import MagicMock from importlib.metadata import distributions from packaging import version from ddtrace import tracer from modular_api.helpers.constants import WINDOWS, LINUX, MODULES_DIR, \ API_MODULE_FILE, MODULE_NAME_KEY, CLI_PATH_KEY, MOUNT_POINT_KEY, \ TOOL_VERSION_MAPPING, DEPENDENCIES, MIN_VER from modular_api.commands_generator import generate_valid_commands from modular_api.helpers.decorators import CommandResponse from modular_api.helpers.exceptions import ModularApiBadRequestException from modular_api.helpers.log_helper import get_logger from modular_api.version import __version__ DESCRIPTOR_REQUIRED_KEYS = (CLI_PATH_KEY, MOUNT_POINT_KEY, MODULE_NAME_KEY) MODULAR_ADMIN_ROOT_PATH = os.path.split(os.path.dirname(__file__))[0] tracer.configure(writer=MagicMock()) _LOG = get_logger(__name__) def install_module_with_destination_folder(paths_to_module: str): """ Installing module by path :param paths_to_module: path to the modules :return: stdout, stderror of the installation process """ if not paths_to_module: message = f"Path not found: {paths_to_module}" _LOG.error(message) sys.exit(message) if os.path.isfile(paths_to_module): message = f'The path {[paths_to_module]} to the module is file. ' \ f'Please specify the path to folder of the module which ' \ f'consist of setup.py.' _LOG.error(message) sys.exit(message) _LOG.info(f"Going to execute pip install command for {paths_to_module}") os_name = os.name command = f'pip install -e {paths_to_module}' if os_name == WINDOWS: with subprocess.Popen( command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) as terminal_process: stdout, stderr = terminal_process.communicate() elif os_name == LINUX: with subprocess.Popen( shlex.split(command), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) as terminal_process: stdout, stderr = terminal_process.communicate() else: message = f'The {os_name} OS is not supported by tool.' _LOG.error(message) sys.exit(message) if stdout is not None: stdout = stdout.decode('utf-8') _LOG.info(f"Out: {stdout}") if stderr is not None: stderr = stderr.decode('utf-8') _LOG.error(f"Errors: {stderr}") _LOG.info('Installation completed with errors') sys.exit(stderr) def write_generated_meta_to_file(path_to_file, mount_point, groups_meta): if not os.path.isfile(path_to_file): cmd_base_content = json.dumps({mount_point: groups_meta}, separators=(',', ':')) else: with open(path_to_file, 'r') as cmd_base: cmd_base_content = json.load(cmd_base) cmd_base_content.update({mount_point: groups_meta}) cmd_base_content = json.dumps(cmd_base_content, separators=(',', ':')) with open(path_to_file, 'w') as cmd_base: cmd_base.write(cmd_base_content) def check_module_requirements(api_module_config): """ Expecting module descriptor file template: { "module_name": "$MODULE_NAME", "cli_path": "$MODULE_CLI_PATH", "mount_point": "$MOUNT_POINT", "dependencies": [ { "module_name": "$MODULE_NAME_DEPENDENT_FROM", "min_version": "$MIN_ALLOWED_VERSION_OF_DEPENDED_MODULE" }, ... ] } Property "dependencies" is optional, but if present - "module_name" in "dependencies"`s objects is required. Property "min_version" is optional """ dependencies = api_module_config.get(DEPENDENCIES) if not dependencies: return candidate = api_module_config.get(MODULE_NAME_KEY) installed_packages = {dist.metadata['Name']: dist for dist in distributions()} for item in dependencies: # check dependent module is installed dependent_module_name = item.get(MODULE_NAME_KEY) if not dependent_module_name: message = 'Missing required property "module_name" in module ' \ 'descriptor file' _LOG.error(message) sys.exit(message) installed_module_name = installed_packages.get(dependent_module_name) if not installed_module_name: message = f'Module "{dependent_module_name}" is marked as ' \ f'required for "{candidate}" module. Please install ' \ f'"{dependent_module_name}" first' _LOG.error(message) sys.exit(message) # check major versions conflict dependency_min_version = item.get(MIN_VER) if not dependency_min_version: break installing_major_min_version = version.parse(dependency_min_version).major installed_major_version = version.parse(installed_module_name.version).major if installing_major_min_version > installed_major_version: message = f'Module "{candidate}" requires a later major version ' \ f'of "{dependent_module_name}". Please update ' \ f'"{dependent_module_name}" to the latest version' _LOG.error(message) sys.exit(message) def extract_module_requirements_setup_py(module_path: str) -> list[str]: """ Extracts the dependencies from the setup.py file :param module_path: path to the setup.py file :return: a list of strings, where each string represents a dependency """ # get dependencies for module to be installed module_dependencies = [] with open(module_path) as module_file: parsed = ast.parse(module_file.read()) for node in parsed.body: if not isinstance(node, ast.Expr): continue if not isinstance(node.value, ast.Call): continue if node.value.func.id != "setup": continue for keyword in node.value.keywords: if keyword.arg == "install_requires": module_dependencies = ast.literal_eval(keyword.value) return module_dependencies def extract_module_requirements_setup_cfg(module_path: str) -> list[str]: """ Extracts the dependencies from a setup.cfg file :param module_path: path to the setup.cfg file :return: a list of strings, where each string represents a dependency """ config = configparser.ConfigParser() config.read(module_path) try: module_dependencies = config.get( 'options', 'install_requires' ).split('\n') module_dependencies = [ dependency for dependency in module_dependencies if dependency ] except (configparser.NoSectionError, configparser.NoOptionError): module_dependencies = [] return module_dependencies def extract_module_requirements_toml(module_path: str) -> list[str]: """ Extracts the dependencies from a pyproject.toml file :param module_path: path to the pyproject.toml file :return: a list of strings, where each string represents a dependency """ # get dependencies for module to be installed with open(module_path, 'r') as module_file: parsed = toml.load(module_file) module_dependencies = parsed['project']['dependencies'] return module_dependencies def check_module_requirements_compatibility( module_dependencies: list[str], module_name: str ): """ Checks if the version requirements for the dependencies are compatible with the modules currently installed in the API's dependency list :param module_dependencies: a list of module dependencies :param module_name: the name of the module whose dependencies are being checked :return: None, but raises SystemExit if a specific version is not provided or there is a version conflict """ # get current dependencies list for Modular-API installed_packages = { dist.metadata['Name']: dist.version for dist in distributions() } modular_api_dependencies = [ f"{name}=={ver}" for name, ver in installed_packages.items() ] # check sticking for a specific version if not module_dependencies: return for req in module_dependencies: if "[" in req: continue # todo refactor - stick to major version if ">=" in req: continue if len(req.split('==')) == 1: message = ( f'Please add a specific version for package \'{req}\' in module' f' \'{module_name}\'' ) _LOG.error(message) sys.exit(message) # check versions compatibility for mod_req in module_dependencies: for api_req in modular_api_dependencies: if "[" in mod_req: continue try: # todo refactor - stick to major version mod_req_name, mod_req_ver = mod_req.split('==') api_req_name, api_req_ver = api_req.split('==') except ValueError: continue if mod_req_name != api_req_name: continue version_to_install = version.parse(mod_req_ver) version_should_be = version.parse(api_req_ver) if version_to_install != version_should_be: message = f'Modular-API has \'{version_should_be}\' version ' \ f'of the \'{api_req_name}\', but module ' \ f'\'{module_name}\' stick to \'{version_to_install}\' ' \ f'version. Please resolve version conflict.' _LOG.error(message) sys.exit(message) @tracer.wrap() def install_module(module_path): """ :param module_path: the path to the installing module :return: none """ extract_dependencies_func_map = { 'setup.py': extract_module_requirements_setup_py, 'setup.cfg': extract_module_requirements_setup_cfg, 'pyproject.toml': extract_module_requirements_toml } setup_files = ["pyproject.toml", "setup.cfg", "setup.py"] _LOG.info(f'Going to install module from path: {module_path}') if not os.path.isdir(module_path): incorrect_path_message = ( 'Provided path is incorrect. It should be a directory.' ) _LOG.error(incorrect_path_message) sys.exit(incorrect_path_message) with open(os.path.join(module_path, API_MODULE_FILE)) as file: api_module_config = json.load(file) _LOG.info('Checking module descriptor properties') if not all( [key in api_module_config.keys() for key in DESCRIPTOR_REQUIRED_KEYS] ): descriptor_key_absence_message = \ f'Descriptor file must contains the following keys: ' \ f'{", ".join(DESCRIPTOR_REQUIRED_KEYS)}' _LOG.error(descriptor_key_absence_message) sys.exit(descriptor_key_absence_message) # Check each setup file by priority for setup_file in setup_files: setup_file_path = os.path.join(module_path, setup_file) if not os.path.isfile(setup_file_path): continue # Skip if file doesn't exist _LOG.info('Checking module requirements compatibility') try: _LOG.info(f"Reading dependencies from: {setup_file}") extract_func = extract_dependencies_func_map[setup_file] module_dependencies = extract_func( module_path=setup_file_path ) except KeyError: _LOG.error(f'Unsupported setup file: {setup_file}') sys.exit(f'Unsupported setup file: {setup_file}') check_module_requirements_compatibility( module_dependencies=module_dependencies, module_name=api_module_config.get('module_name') ) # Found valid setup file, break the loop _LOG.info(f"Successfully loaded dependencies from {setup_file}") break else: _LOG.error("No valid setup file found") sys.exit("No valid setup file found") _LOG.info('Checking module dependencies') check_module_requirements(api_module_config) modular_admin_path, _ = os.path.split(os.path.dirname(__file__)) destination_folder = os.path.join(modular_admin_path, MODULES_DIR, api_module_config[MODULE_NAME_KEY]) module_name = api_module_config[MODULE_NAME_KEY] path_to_module = os.path.join(MODULAR_ADMIN_ROOT_PATH, MODULES_DIR, module_name) if os.path.exists(path_to_module): _LOG.warning(f'The \'{module_name}\' module will be reinstalled') remove_tree(path_to_module) copytree( module_path, destination_folder, ignore=ignore_patterns( '*.tox', 'build', '*.egg-info', '*.git', 'tests', 'requirements-dev.txt', 'tox.ini', 'logs') ) install_module_with_destination_folder(paths_to_module=destination_folder) _LOG.info(f'Copy {api_module_config[MODULE_NAME_KEY]} module files ' f'to {destination_folder}') mount_point = api_module_config[MOUNT_POINT_KEY] valid_methods = generate_valid_commands( destination_folder=destination_folder, path_to_scan=os.path.join(modular_admin_path, MODULES_DIR, api_module_config[MODULE_NAME_KEY], *api_module_config[CLI_PATH_KEY].split('/')), mount_point=mount_point, is_private_mode_enabled=False, path_to_setup_file_in_module=setup_file_path ) web_service_cmd_base = os.path.join(modular_admin_path, 'web_service', 'commands_base.json') _LOG.info(f'Updating commands meta file by path: {web_service_cmd_base}') write_generated_meta_to_file(path_to_file=web_service_cmd_base, mount_point=mount_point, groups_meta=valid_methods) message = f'{api_module_config[MODULE_NAME_KEY].capitalize()} ' \ f'successfully installed' _LOG.info(message) return CommandResponse(message=message) def check_uninstall(api_module_config): """ Expecting module descriptor file template: { "module_name": "$MODULE_NAME", "cli_path": "$MODULE_CLI_PATH", "mount_point": "$MOUNT_POINT", "dependencies": [ { "module_name": "$MODULE_NAME_DEPENDENT_FROM", }, ... ] } Property "dependencies" is optional, but if present - "module_name" in "dependencies"`s objects is required. """ modules_path = Path(__file__).parent.parent / MODULES_DIR if not modules_path.exists(): return conflict_modules_list = list() for module in modules_path.iterdir(): api_file_path = module / API_MODULE_FILE if not api_file_path.exists(): continue with open(api_file_path, 'r') as file: api_module_file = json.load(file) dependencies = list() if not api_module_file.get(DEPENDENCIES): continue for item in api_module_file.get(DEPENDENCIES): dependencies.append(item.get(MODULE_NAME_KEY)) conflict_modules_list.append({ f"{api_module_file.get(MODULE_NAME_KEY)}": dependencies }) if not conflict_modules_list: return uninstall_candidate = api_module_config.get(MODULE_NAME_KEY) result_list = list() for conflict_item in conflict_modules_list: for k, v in conflict_item.items(): if uninstall_candidate in v: result_list.append(k) if result_list: message = f'Module "{uninstall_candidate}" you are trying to ' \ f'uninstall is in dependencies of the another module(s). ' \ f'Please uninstall "{", ".join(result_list)}" module(s) first' _LOG.error(message) sys.exit(message) @tracer.wrap() def uninstall_module(module_name): """ :param module_name: the name to the module to uninstalling :return: none """ _LOG.info(f"Going to delete the '{module_name}' module") m3_modular_admin_dir, _ = os.path.split(os.path.dirname(__file__)) module_descriptor_path = os.path.join( m3_modular_admin_dir, MODULES_DIR, module_name, API_MODULE_FILE) if not os.path.isfile(module_descriptor_path): incorrect_path_message = 'Provided path is incorrect or does ' \ 'not contain api_module.json file' _LOG.error(incorrect_path_message) raise ModularApiBadRequestException(incorrect_path_message) with open(module_descriptor_path) as file: api_module_config = json.load(file) check_uninstall(api_module_config) if not all(key in api_module_config.keys() for key in DESCRIPTOR_REQUIRED_KEYS): descriptor_key_absence_message = \ f'Descriptor file must contains the following keys: ' \ f'{", ".join(DESCRIPTOR_REQUIRED_KEYS)}' _LOG.error(descriptor_key_absence_message) sys.exit(descriptor_key_absence_message) web_service_cmd_base = os.path.join(MODULAR_ADMIN_ROOT_PATH, 'web_service', 'commands_base.json') mount_point = api_module_config[MOUNT_POINT_KEY] with open(web_service_cmd_base) as file: web_service_content = json.load(file) _LOG.info(f'Deleting the {mount_point} mount point from metadata') web_service_content.pop(mount_point, None) with open(web_service_cmd_base, 'w') as file: json.dump(web_service_content, file, indent=2) remove_tree( os.path.join(MODULAR_ADMIN_ROOT_PATH, MODULES_DIR, module_name)) _LOG.info(f'The {module_name} module was successfully uninstalled') return CommandResponse( message=f'\'{module_name}\' successfully uninstalled') def check_and_describe_modules( table_response: bool | None = None, json_response: bool | None = None, ) -> CommandResponse: if table_response and json_response: _LOG.error('Wrong parameters passed') raise ModularApiBadRequestException( 'Please specify only one parameter - table or json') modules_path = Path(__file__).parent.parent / MODULES_DIR if not modules_path.exists(): _LOG.warning(f'Directory \'{MODULES_DIR}\' does not exist') return CommandResponse( message=f'Missing \'{MODULES_DIR}\' by path \'{str(modules_path)}\'.' f' Nothing to describe, please install any module first' ) installed_modules_list = [] for module in modules_path.iterdir(): api_file_path = module / API_MODULE_FILE if not module.is_dir() or not api_file_path.exists(): continue with open(api_file_path, 'r'): installed_modules_list.append(module.name) if not installed_modules_list: _LOG.warning('Modules are not installed') return CommandResponse( message='Can not find any installed module, nothing to describe' ) modular_sdk_version = 'Modular-SDK: {0}' modular_cli_sdk_version = 'Modular-CLI-SDK: {0}' result_message = 'Installed modules:' pretty_table = list() installed_packages = {dist.metadata['Name']: dist.version for dist in distributions()} installed_packages_list = sorted( ["%s@%s" % (name, version) for name, version in installed_packages.items()] ) for module_name in installed_modules_list: for package_name in installed_packages_list: item_name, item_version = package_name.split('@') if item_name == 'modular_sdk': modular_sdk_version = modular_sdk_version.format(item_version) if item_name == 'modular-cli-sdk': modular_cli_sdk_version = modular_cli_sdk_version.format(item_version) if item_name == module_name: pretty_table.append( {"Module name": module_name, "Version": item_version} ) result_message += TOOL_VERSION_MAPPING.format( tool=module_name.ljust(15), version=item_version) if modular_sdk_version == 'Modular-SDK: {0}': modular_sdk_version = 'Modular-SDK: Not installed' if modular_cli_sdk_version == 'Modular-CLI-SDK: {0}': modular_cli_sdk_version = 'Modular-CLI-SDK: Not installed' modular_version = f'Modular-API: {__version__}' if json_response: modular_sdk_item = modular_sdk_version.split(':')[0].lower() modular_sdk_item_ver = modular_sdk_version.split(':')[1].lower() modular_cli_sdk_item = modular_cli_sdk_version.split(':')[0].lower() modular_cli_sdk_item_ver = modular_cli_sdk_version.split(':')[1].lower() result_json = { 'modular': __version__, modular_sdk_item: modular_sdk_item_ver.strip(), modular_cli_sdk_item: modular_cli_sdk_item_ver.strip() } for item in pretty_table: item_name = item.get('Module name') item_vers = item.get('Version') result_json.update({item_name: item_vers}) return CommandResponse( message=json.dumps(result_json, indent=4) ) if table_response: return CommandResponse( table_title=modular_version + '\n' + modular_sdk_version + '\n' + modular_cli_sdk_version + '\n' + 'Installed modules', items=pretty_table) return CommandResponse( message=modular_version + '\n' + modular_sdk_version + '\n' + modular_cli_sdk_version + '\n' + result_message)