import argparse import os import textwrap import re import shutil from collections import defaultdict from contextlib import suppress from pathlib import Path from typing import List, Optional, Tuple from httpie.manager.compat import PipError, run_pip from httpie.manager.cli import parser, missing_subcommand from httpie.compat import get_dist_name, importlib_metadata from httpie.context import Environment from httpie.status import ExitStatus from httpie.utils import get_site_paths PEP_503 = re.compile(r"[-_.]+") class PluginInstaller: def __init__(self, env: Environment, debug: bool = False) -> None: self.env = env self.dir = env.config.plugins_dir self.debug = debug self.setup_plugins_dir() def setup_plugins_dir(self) -> None: try: self.dir.mkdir( exist_ok=True, parents=True ) except OSError: self.env.stderr.write( f'Couldn\'t create "{self.dir!s}"' ' directory for plugin installation.' ' Please re-check the permissions for that directory,' ' and if needed, allow write-access.' ) raise def fail( self, command: str, target: Optional[str] = None, reason: Optional[str] = None ) -> ExitStatus: message = f'Can\'t {command}' if target: message += f' {target!r}' if reason: message += f': {reason}' self.env.stderr.write(message + '\n') return ExitStatus.ERROR def _install(self, targets: List[str], mode='install') -> Tuple[ bytes, ExitStatus ]: pip_args = [ 'install', '--prefer-binary', f'--prefix={self.dir}', '--no-warn-script-location', ] if mode == 'upgrade': pip_args.append('--upgrade') pip_args.extend(targets) try: stdout = run_pip(pip_args) except PipError as pip_error: error = pip_error stdout = pip_error.stdout else: error = None self.env.stdout.write(stdout.decode()) if error: reason = None if error.stderr: stderr = error.stderr.decode() if self.debug: self.env.stderr.write('Command failed: ') self.env.stderr.write('pip ' + ' '.join(pip_args) + '\n') self.env.stderr.write(textwrap.indent(' ', stderr)) last_line = stderr.strip().splitlines()[-1] severity, _, message = last_line.partition(': ') if severity == 'ERROR': reason = message stdout = error.stdout exit_status = self.fail(mode, ', '.join(targets), reason) else: exit_status = ExitStatus.SUCCESS return stdout, exit_status def install(self, targets: List[str]) -> ExitStatus: self.env.stdout.write(f"Installing {', '.join(targets)}...\n") self.env.stdout.flush() _, exit_status = self._install(targets) return exit_status def _clear_metadata(self, targets: List[str]) -> None: # Due to an outstanding pip problem[0], we have to get rid of # existing metadata for old versions manually. # [0]: https://github.com/pypa/pip/issues/10727 result_deps = defaultdict(list) for site_dir in get_site_paths(self.dir): for child in site_dir.iterdir(): if child.suffix in {'.dist-info', '.egg-info'}: name, _, version = child.stem.rpartition('-') result_deps[name].append((version, child)) for target in targets: name, _, version = target.rpartition('-') name = PEP_503.sub("-", name).lower().replace('-', '_') if name not in result_deps: continue for result_version, meta_path in result_deps[name]: if version != result_version: shutil.rmtree(meta_path) def upgrade(self, targets: List[str]) -> ExitStatus: self.env.stdout.write(f"Upgrading {', '.join(targets)}...\n") self.env.stdout.flush() raw_stdout, exit_status = self._install( targets, mode='upgrade' ) if not raw_stdout: return exit_status stdout = raw_stdout.decode() installation_line = stdout.splitlines()[-1] if installation_line.startswith('Successfully installed'): self._clear_metadata(installation_line.split()[2:]) def _uninstall(self, target: str) -> Optional[ExitStatus]: try: distribution = importlib_metadata.distribution(target) except importlib_metadata.PackageNotFoundError: return self.fail('uninstall', target, 'package is not installed') base_dir = Path(distribution.locate_file('.')).resolve() if self.dir not in base_dir.parents: # If the package is installed somewhere else (e.g on the site packages # of the real python interpreter), than that means this package is not # installed through us. return self.fail('uninstall', target, 'package is not installed through httpie plugins' ' interface') files = distribution.files if files is None: return self.fail('uninstall', target, 'couldn\'t locate the package') # TODO: Consider handling failures here (e.g if it fails, # just revert the operation and leave the site-packages # in a proper shape). for file in files: with suppress(FileNotFoundError): os.unlink(distribution.locate_file(file)) metadata_path = getattr(distribution, '_path', None) if ( metadata_path and metadata_path.exists() and not any(metadata_path.iterdir()) ): metadata_path.rmdir() self.env.stdout.write(f'Successfully uninstalled {target}\n') def uninstall(self, targets: List[str]) -> ExitStatus: # Unfortunately uninstall doesn't work with custom pip schemes. See: # - https://github.com/pypa/pip/issues/5595 # - https://github.com/pypa/pip/issues/4575 # so we have to implement our own uninstalling logic. Which works # on top of the importlib_metadata. exit_code = ExitStatus.SUCCESS for target in targets: exit_code |= self._uninstall(target) or ExitStatus.SUCCESS return ExitStatus(exit_code) def list(self) -> None: from httpie.plugins.registry import plugin_manager known_plugins = defaultdict(list) for entry_point in plugin_manager.iter_entry_points(self.dir): ep_info = (entry_point.group, entry_point.name) ep_name = get_dist_name(entry_point) or entry_point.module known_plugins[ep_name].append(ep_info) for plugin, entry_points in known_plugins.items(): self.env.stdout.write(plugin) version = importlib_metadata.version(plugin) if version is not None: self.env.stdout.write(f' ({version})') self.env.stdout.write('\n') for group, entry_point in sorted(entry_points): self.env.stdout.write(f' {entry_point} ({group})\n') def run( self, action: Optional[str], args: argparse.Namespace, ) -> ExitStatus: from httpie.plugins.manager import enable_plugins if action is None: parser.error(missing_subcommand('plugins')) with enable_plugins(self.dir): if action == 'install': status = self.install(args.targets) elif action == 'upgrade': status = self.upgrade(args.targets) elif action == 'uninstall': status = self.uninstall(args.targets) elif action == 'list': status = self.list() return status or ExitStatus.SUCCESS def cli_plugins(env: Environment, args: argparse.Namespace) -> ExitStatus: plugins = PluginInstaller(env, debug=args.debug) try: action = args.cli_plugins_action except AttributeError: action = args.plugins_action return plugins.run(action, args)