def install_module()

in modular_api/services/install_service.py [0:0]
89 lines of code
7 McCabe index (conditional complexity)

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)