codekingpro/portable-devtools

1import argparse2import os3import textwrap4import re5import shutil6from collections import defaultdict7from contextlib import suppress8from pathlib import Path9from typing import List, Optional, Tuple10 11from httpie.manager.compat import PipError, run_pip12from httpie.manager.cli import parser, missing_subcommand13from httpie.compat import get_dist_name, importlib_metadata14from httpie.context import Environment15from httpie.status import ExitStatus16from httpie.utils import get_site_paths17 18PEP_503 = re.compile(r"[-_.]+")19 20 21class PluginInstaller:22 23    def __init__(self, env: Environment, debug: bool = False) -> None:24        self.env = env25        self.dir = env.config.plugins_dir26        self.debug = debug27 28        self.setup_plugins_dir()29 30    def setup_plugins_dir(self) -> None:31        try:32            self.dir.mkdir(33                exist_ok=True,34                parents=True35            )36        except OSError:37            self.env.stderr.write(38                f'Couldn\'t create "{self.dir!s}"'39                ' directory for plugin installation.'40                ' Please re-check the permissions for that directory,'41                ' and if needed, allow write-access.'42            )43            raise44 45    def fail(46        self,47        command: str,48        target: Optional[str] = None,49        reason: Optional[str] = None50    ) -> ExitStatus:51        message = f'Can\'t {command}'52        if target:53            message += f' {target!r}'54        if reason:55            message += f': {reason}'56 57        self.env.stderr.write(message + '\n')58        return ExitStatus.ERROR59 60    def _install(self, targets: List[str], mode='install') -> Tuple[61        bytes, ExitStatus62    ]:63        pip_args = [64            'install',65            '--prefer-binary',66            f'--prefix={self.dir}',67            '--no-warn-script-location',68        ]69        if mode == 'upgrade':70            pip_args.append('--upgrade')71        pip_args.extend(targets)72 73        try:74            stdout = run_pip(pip_args)75        except PipError as pip_error:76            error = pip_error77            stdout = pip_error.stdout78        else:79            error = None80 81        self.env.stdout.write(stdout.decode())82 83        if error:84            reason = None85            if error.stderr:86                stderr = error.stderr.decode()87 88                if self.debug:89                    self.env.stderr.write('Command failed: ')90                    self.env.stderr.write('pip ' + ' '.join(pip_args) + '\n')91                    self.env.stderr.write(textwrap.indent('  ', stderr))92 93                last_line = stderr.strip().splitlines()[-1]94                severity, _, message = last_line.partition(': ')95                if severity == 'ERROR':96                    reason = message97 98            stdout = error.stdout99            exit_status = self.fail(mode, ', '.join(targets), reason)100        else:101            exit_status = ExitStatus.SUCCESS102 103        return stdout, exit_status104 105    def install(self, targets: List[str]) -> ExitStatus:106        self.env.stdout.write(f"Installing {', '.join(targets)}...\n")107        self.env.stdout.flush()108        _, exit_status = self._install(targets)109        return exit_status110 111    def _clear_metadata(self, targets: List[str]) -> None:112        # Due to an outstanding pip problem[0], we have to get rid of113        # existing metadata for old versions manually.114        # [0]: https://github.com/pypa/pip/issues/10727115        result_deps = defaultdict(list)116        for site_dir in get_site_paths(self.dir):117            for child in site_dir.iterdir():118                if child.suffix in {'.dist-info', '.egg-info'}:119                    name, _, version = child.stem.rpartition('-')120                    result_deps[name].append((version, child))121 122        for target in targets:123            name, _, version = target.rpartition('-')124            name = PEP_503.sub("-", name).lower().replace('-', '_')125            if name not in result_deps:126                continue127 128            for result_version, meta_path in result_deps[name]:129                if version != result_version:130                    shutil.rmtree(meta_path)131 132    def upgrade(self, targets: List[str]) -> ExitStatus:133        self.env.stdout.write(f"Upgrading {', '.join(targets)}...\n")134        self.env.stdout.flush()135 136        raw_stdout, exit_status = self._install(137            targets,138            mode='upgrade'139        )140        if not raw_stdout:141            return exit_status142 143        stdout = raw_stdout.decode()144        installation_line = stdout.splitlines()[-1]145        if installation_line.startswith('Successfully installed'):146            self._clear_metadata(installation_line.split()[2:])147 148    def _uninstall(self, target: str) -> Optional[ExitStatus]:149        try:150            distribution = importlib_metadata.distribution(target)151        except importlib_metadata.PackageNotFoundError:152            return self.fail('uninstall', target, 'package is not installed')153 154        base_dir = Path(distribution.locate_file('.')).resolve()155        if self.dir not in base_dir.parents:156            # If the package is installed somewhere else (e.g on the site packages157            # of the real python interpreter), than that means this package is not158            # installed through us.159            return self.fail('uninstall', target,160                             'package is not installed through httpie plugins'161                             ' interface')162 163        files = distribution.files164        if files is None:165            return self.fail('uninstall', target, 'couldn\'t locate the package')166 167        # TODO: Consider handling failures here (e.g if it fails,168        # just revert the operation and leave the site-packages169        # in a proper shape).170        for file in files:171            with suppress(FileNotFoundError):172                os.unlink(distribution.locate_file(file))173 174        metadata_path = getattr(distribution, '_path', None)175        if (176            metadata_path177            and metadata_path.exists()178            and not any(metadata_path.iterdir())179        ):180            metadata_path.rmdir()181 182        self.env.stdout.write(f'Successfully uninstalled {target}\n')183 184    def uninstall(self, targets: List[str]) -> ExitStatus:185        # Unfortunately uninstall doesn't work with custom pip schemes. See:186        # - https://github.com/pypa/pip/issues/5595187        # - https://github.com/pypa/pip/issues/4575188        # so we have to implement our own uninstalling logic. Which works189        # on top of the importlib_metadata.190 191        exit_code = ExitStatus.SUCCESS192        for target in targets:193            exit_code |= self._uninstall(target) or ExitStatus.SUCCESS194        return ExitStatus(exit_code)195 196    def list(self) -> None:197        from httpie.plugins.registry import plugin_manager198 199        known_plugins = defaultdict(list)200 201        for entry_point in plugin_manager.iter_entry_points(self.dir):202            ep_info = (entry_point.group, entry_point.name)203            ep_name = get_dist_name(entry_point) or entry_point.module204            known_plugins[ep_name].append(ep_info)205 206        for plugin, entry_points in known_plugins.items():207            self.env.stdout.write(plugin)208 209            version = importlib_metadata.version(plugin)210            if version is not None:211                self.env.stdout.write(f' ({version})')212            self.env.stdout.write('\n')213 214            for group, entry_point in sorted(entry_points):215                self.env.stdout.write(f'  {entry_point} ({group})\n')216 217    def run(218        self,219        action: Optional[str],220        args: argparse.Namespace,221    ) -> ExitStatus:222        from httpie.plugins.manager import enable_plugins223 224        if action is None:225            parser.error(missing_subcommand('plugins'))226 227        with enable_plugins(self.dir):228            if action == 'install':229                status = self.install(args.targets)230            elif action == 'upgrade':231                status = self.upgrade(args.targets)232            elif action == 'uninstall':233                status = self.uninstall(args.targets)234            elif action == 'list':235                status = self.list()236 237        return status or ExitStatus.SUCCESS238 239 240def cli_plugins(env: Environment, args: argparse.Namespace) -> ExitStatus:241    plugins = PluginInstaller(env, debug=args.debug)242 243    try:244        action = args.cli_plugins_action245    except AttributeError:246        action = args.plugins_action247 248    return plugins.run(action, args)249