install_cmd.py
1 import tempfile 2 import os 3 import re 4 import subprocess 5 import pathlib 6 import click 7 import shutil 8 import toml 9 10 from cli.utils import get_module_path, error_exit 11 from .official_registry import get_official_plugin_url 12 13 14 def _check_command_exists(command: str) -> bool: 15 """Checks if a command exists on the system.""" 16 return shutil.which(command) is not None 17 18 def _get_plugin_name_from_source_pyproject(source_path: pathlib.Path) -> str | None: 19 """Reads pyproject.toml from source_path and returns the project name.""" 20 pyproject_path = source_path / "pyproject.toml" 21 if not pyproject_path.is_file(): 22 click.echo( 23 click.style( 24 f"Warning: pyproject.toml not found at {pyproject_path}. Cannot determine module name automatically.", 25 fg="yellow", 26 ) 27 ) 28 return None 29 try: 30 with open(pyproject_path, "r", encoding="utf-8") as f: 31 data = toml.load(f) 32 project_name = data.get("project", {}).get("name") 33 if project_name: 34 return project_name.strip().replace("-", "_") # Normalize to snake_case 35 click.echo( 36 click.style( 37 f"Warning: Could not find 'project.name' in {pyproject_path}.", 38 fg="yellow", 39 ) 40 ) 41 return None 42 except Exception as e: 43 click.echo(click.style(f"Error parsing {pyproject_path}: {e}", fg="red")) 44 return None 45 46 def _run_install( 47 install_command, install_target: str | pathlib.Path, operation_desc: str 48 ) -> str | None: 49 """Runs install for the given target.""" 50 click.echo( 51 f"Attempting to install plugin using {install_command} from {operation_desc}..." 52 ) 53 try: 54 process = subprocess.run( 55 install_command.format(package=str(install_target)).split(), 56 capture_output=True, 57 text=True, 58 check=False, 59 ) 60 if process.returncode == 0: 61 click.echo( 62 click.style( 63 f"Plugin successfully installed via from {operation_desc}.", 64 fg="green", 65 ) 66 ) 67 if process.stdout: 68 click.echo(f"install output:\n{process.stdout}") 69 return None 70 else: 71 return f"Error: 'install {install_target}' failed.\nstdout:\n{process.stdout}\nstderr:\n{process.stderr}" 72 except FileNotFoundError: 73 return "Error: 'python' or command not found. Ensure Python and command are installed and in your PATH." 74 except Exception as e: 75 return f"An unexpected error occurred during install: {e}" 76 77 def install_plugin(plugin_source: str, installer_command: str | None = None) -> tuple[str | None, pathlib.Path | None]: 78 """Installs a plugin from the specified source. 79 80 Args: 81 plugin_source: Source of the plugin (module name, local path, or Git URL) 82 installer_command: Command to install the plugin, with '{package}' placeholder 83 Returns: 84 Tuple of (module_name, plugin_path) if successful, or (None, None) on failure 85 """ 86 if not installer_command: 87 installer_command = os.environ.get( 88 "SAM_PLUGIN_INSTALL_COMMAND", "pip3 install {package}" 89 ) 90 try: 91 installer_command.format(package="dummy") # Test if the command is valid 92 except (KeyError, ValueError): 93 return error_exit( 94 "Error: The installer command must contain a placeholder '{package}' to be replaced with the actual package name." 95 ) 96 97 official_plugin_url = get_official_plugin_url(plugin_source) 98 if official_plugin_url: 99 click.echo(f"Found official plugin '{plugin_source}' at: {official_plugin_url}") 100 plugin_source = official_plugin_url 101 102 install_type = None # "module", "local", "git", "pypi" 103 module_name = None 104 install_target = None 105 source_path_for_name_extraction = None 106 107 # If the resolved official URL is a plain package name (not a URL), install from PyPI 108 if official_plugin_url and not plugin_source.startswith( 109 ("git+", "http://", "https://", "file://", "/", "./", "../", "~/") 110 ): 111 install_type = "pypi" 112 install_target = plugin_source # pip-style dashed name 113 module_name = plugin_source.strip() 114 115 elif plugin_source.startswith(("http://", "https://")) and plugin_source.endswith( 116 ".git" 117 ): 118 install_type = "repository" 119 install_target = plugin_source 120 elif plugin_source.startswith(("git+")): 121 install_type = "git" 122 install_target = plugin_source 123 elif os.path.exists(plugin_source): 124 local_path = pathlib.Path(plugin_source).resolve() 125 if local_path.is_dir(): 126 install_type = "local" 127 install_target = str(local_path) 128 source_path_for_name_extraction = local_path 129 elif local_path.is_file() and local_path.suffix in [".whl", ".tar.gz"]: 130 install_type = "wheel" 131 install_target = str(local_path) 132 else: 133 return error_exit( 134 f"Error: Local path '{plugin_source}' exists but is not a directory or wheel." 135 ) 136 elif not re.search(r"[/\\]", plugin_source): 137 install_type = "module" 138 module_name = plugin_source.strip().replace("-", "_") 139 else: 140 return error_exit( 141 f"Error: Invalid plugin source '{plugin_source}'. Not a recognized module name, local path, or Git URL." 142 ) 143 144 if install_type in ["local", "git", "repository", "wheel", "pypi"]: 145 if install_type == "repository": 146 if not _check_command_exists("git"): 147 return error_exit( 148 "Error: 'git' command not found. Please install Git or install the plugin manually." 149 ) 150 151 with tempfile.TemporaryDirectory() as temp_dir: 152 temp_dir_path = pathlib.Path(temp_dir) 153 cloned_repo_path = temp_dir_path / "plugin_repo" 154 click.echo( 155 f"Cloning Git repository '{plugin_source}' to temporary directory {cloned_repo_path}..." 156 ) 157 try: 158 subprocess.run( 159 ["git", "clone", plugin_source, str(cloned_repo_path)], 160 capture_output=True, 161 text=True, 162 check=True, 163 ) 164 source_path_for_name_extraction = cloned_repo_path 165 except subprocess.CalledProcessError as e: 166 return error_exit(f"Error cloning Git repository: {e.stderr}") 167 except FileNotFoundError: 168 return error_exit("Error: 'git' command not found during clone.") 169 170 module_name_from_pyproject = _get_plugin_name_from_source_pyproject( 171 source_path_for_name_extraction 172 ) 173 if not module_name_from_pyproject: 174 return error_exit( 175 "Could not determine module name from pyproject.toml in the Git repo. Aborting." 176 ) 177 178 err = _run_install( 179 installer_command, install_target, f"Git URL ({plugin_source})" 180 ) 181 if err: 182 return error_exit(err) 183 module_name = module_name_from_pyproject 184 185 elif install_type == "git": 186 module_name_from_url = ( 187 plugin_source.split("#")[0] 188 .split("?")[0] 189 .split("/")[-1] 190 .replace(".git", "") 191 .replace("-", "_") 192 ) 193 if "#subdirectory=" in plugin_source: 194 module_name_from_url = ( 195 plugin_source.split("#subdirectory=")[-1] 196 .split("?")[0] 197 .replace(".git", "") 198 .replace("-", "_") 199 ) 200 201 if not module_name_from_url: 202 return error_exit( 203 f"Could not determine module name from the Git URL {plugin_source}. Aborting." 204 ) 205 206 err = _run_install( 207 installer_command, install_target, f"Git URL ({plugin_source})" 208 ) 209 if err: 210 return error_exit(err) 211 module_name = module_name_from_url 212 213 elif install_type == "local": 214 module_name_from_pyproject = _get_plugin_name_from_source_pyproject( 215 source_path_for_name_extraction 216 ) 217 if not module_name_from_pyproject: 218 return error_exit( 219 f"Could not determine module name from pyproject.toml at {source_path_for_name_extraction}. Aborting." 220 ) 221 222 err = _run_install( 223 installer_command, install_target, f"local path ({install_target})" 224 ) 225 if err: 226 return error_exit(err) 227 module_name = module_name_from_pyproject 228 229 elif install_type == "wheel": 230 module_name_from_wheel = ( 231 pathlib.Path(install_target).stem.split("-")[0] 232 ) 233 if not module_name_from_wheel: 234 return error_exit( 235 f"Could not determine module name from the wheel file {install_target}. Aborting." 236 ) 237 238 err = _run_install( 239 installer_command, install_target, f"wheel file ({install_target})" 240 ) 241 if err: 242 return error_exit(err) 243 module_name = module_name_from_wheel 244 245 elif install_type == "pypi": 246 err = _run_install( 247 installer_command, install_target, f"PyPI ({install_target})" 248 ) 249 if err: 250 return error_exit(err) 251 # module_name already set when install_type was determined 252 253 if not module_name: 254 return error_exit("Error: Could not determine the plugin module name to load.") 255 256 click.echo(f"Proceeding to load plugin module '{module_name}'...") 257 try: 258 plugin_path = pathlib.Path(get_module_path(module_name)) 259 except ImportError: 260 return error_exit( 261 f"Error: Plugin module '{module_name}' not found after potential installation. Please check installation logs or install manually." 262 ) 263 264 if not plugin_path or not plugin_path.exists(): 265 return error_exit( 266 f"Error: Could not determine a valid root path for plugin module '{module_name}'. Path: {plugin_path}" 267 ) 268 269 return module_name, plugin_path 270 271 272 @click.command("install") 273 @click.argument("plugin_source") 274 @click.option( 275 "--install-command", 276 "installer_command", 277 help="Command to use to install a python package. Must follow the format 'command {package} args', by default 'pip3 install {package}'. Can also be set through the environment variable SAM_PLUGIN_INSTALL_COMMAND.", 278 ) 279 def install_plugin_cmd( 280 plugin_source: str, 281 installer_command: str | None = None 282 ): 283 """ 284 Installs a plugin from the specified source. 285 286 PLUGIN_SOURCE can be: \n 287 - A local path to a directory (e.g., '/path/to/plugin') \n 288 - A local path to a wheel file (e.g., '/path/to/plugin.whl') \n 289 - A Git URL (e.g., 'https://github.com/user/repo.git') \n 290 - The name of an official plugin (e.g., 'sam-rest-gateway') — installed from PyPI if published, otherwise from the official GitHub registry \n 291 """ 292 module_name, plugin_path = install_plugin(plugin_source, installer_command) 293 if module_name and plugin_path: 294 click.echo( 295 click.style( 296 f"Plugin '{module_name}' installed and available at {plugin_path}.", 297 fg="green", 298 ) 299 )