| from __future__ import absolute_import |
| |
| import json |
| import logging |
| |
| from pip._vendor import six |
| |
| from pip._internal.cli import cmdoptions |
| from pip._internal.cli.req_command import IndexGroupCommand |
| from pip._internal.cli.status_codes import SUCCESS |
| from pip._internal.exceptions import CommandError |
| from pip._internal.index.collector import LinkCollector |
| from pip._internal.index.package_finder import PackageFinder |
| from pip._internal.models.selection_prefs import SelectionPreferences |
| from pip._internal.utils.misc import ( |
| dist_is_editable, |
| get_installed_distributions, |
| tabulate, |
| write_output, |
| ) |
| from pip._internal.utils.packaging import get_installer |
| from pip._internal.utils.parallel import map_multithread |
| from pip._internal.utils.typing import MYPY_CHECK_RUNNING |
| |
| if MYPY_CHECK_RUNNING: |
| from optparse import Values |
| from typing import List, Set, Tuple, Iterator |
| |
| from pip._internal.network.session import PipSession |
| from pip._vendor.pkg_resources import Distribution |
| |
| logger = logging.getLogger(__name__) |
| |
| |
| class ListCommand(IndexGroupCommand): |
| """ |
| List installed packages, including editables. |
| |
| Packages are listed in a case-insensitive sorted order. |
| """ |
| |
| ignore_require_venv = True |
| usage = """ |
| %prog [options]""" |
| |
| def add_options(self): |
| # type: () -> None |
| self.cmd_opts.add_option( |
| '-o', '--outdated', |
| action='store_true', |
| default=False, |
| help='List outdated packages') |
| self.cmd_opts.add_option( |
| '-u', '--uptodate', |
| action='store_true', |
| default=False, |
| help='List uptodate packages') |
| self.cmd_opts.add_option( |
| '-e', '--editable', |
| action='store_true', |
| default=False, |
| help='List editable projects.') |
| self.cmd_opts.add_option( |
| '-l', '--local', |
| action='store_true', |
| default=False, |
| help=('If in a virtualenv that has global access, do not list ' |
| 'globally-installed packages.'), |
| ) |
| self.cmd_opts.add_option( |
| '--user', |
| dest='user', |
| action='store_true', |
| default=False, |
| help='Only output packages installed in user-site.') |
| self.cmd_opts.add_option(cmdoptions.list_path()) |
| self.cmd_opts.add_option( |
| '--pre', |
| action='store_true', |
| default=False, |
| help=("Include pre-release and development versions. By default, " |
| "pip only finds stable versions."), |
| ) |
| |
| self.cmd_opts.add_option( |
| '--format', |
| action='store', |
| dest='list_format', |
| default="columns", |
| choices=('columns', 'freeze', 'json'), |
| help="Select the output format among: columns (default), freeze, " |
| "or json", |
| ) |
| |
| self.cmd_opts.add_option( |
| '--not-required', |
| action='store_true', |
| dest='not_required', |
| help="List packages that are not dependencies of " |
| "installed packages.", |
| ) |
| |
| self.cmd_opts.add_option( |
| '--exclude-editable', |
| action='store_false', |
| dest='include_editable', |
| help='Exclude editable package from output.', |
| ) |
| self.cmd_opts.add_option( |
| '--include-editable', |
| action='store_true', |
| dest='include_editable', |
| help='Include editable package from output.', |
| default=True, |
| ) |
| index_opts = cmdoptions.make_option_group( |
| cmdoptions.index_group, self.parser |
| ) |
| |
| self.parser.insert_option_group(0, index_opts) |
| self.parser.insert_option_group(0, self.cmd_opts) |
| |
| def _build_package_finder(self, options, session): |
| # type: (Values, PipSession) -> PackageFinder |
| """ |
| Create a package finder appropriate to this list command. |
| """ |
| link_collector = LinkCollector.create(session, options=options) |
| |
| # Pass allow_yanked=False to ignore yanked versions. |
| selection_prefs = SelectionPreferences( |
| allow_yanked=False, |
| allow_all_prereleases=options.pre, |
| ) |
| |
| return PackageFinder.create( |
| link_collector=link_collector, |
| selection_prefs=selection_prefs, |
| ) |
| |
| def run(self, options, args): |
| # type: (Values, List[str]) -> int |
| if options.outdated and options.uptodate: |
| raise CommandError( |
| "Options --outdated and --uptodate cannot be combined.") |
| |
| cmdoptions.check_list_path_option(options) |
| |
| packages = get_installed_distributions( |
| local_only=options.local, |
| user_only=options.user, |
| editables_only=options.editable, |
| include_editables=options.include_editable, |
| paths=options.path, |
| ) |
| |
| # get_not_required must be called firstly in order to find and |
| # filter out all dependencies correctly. Otherwise a package |
| # can't be identified as requirement because some parent packages |
| # could be filtered out before. |
| if options.not_required: |
| packages = self.get_not_required(packages, options) |
| |
| if options.outdated: |
| packages = self.get_outdated(packages, options) |
| elif options.uptodate: |
| packages = self.get_uptodate(packages, options) |
| |
| self.output_package_listing(packages, options) |
| return SUCCESS |
| |
| def get_outdated(self, packages, options): |
| # type: (List[Distribution], Values) -> List[Distribution] |
| return [ |
| dist for dist in self.iter_packages_latest_infos(packages, options) |
| if dist.latest_version > dist.parsed_version |
| ] |
| |
| def get_uptodate(self, packages, options): |
| # type: (List[Distribution], Values) -> List[Distribution] |
| return [ |
| dist for dist in self.iter_packages_latest_infos(packages, options) |
| if dist.latest_version == dist.parsed_version |
| ] |
| |
| def get_not_required(self, packages, options): |
| # type: (List[Distribution], Values) -> List[Distribution] |
| dep_keys = set() # type: Set[Distribution] |
| for dist in packages: |
| dep_keys.update(requirement.key for requirement in dist.requires()) |
| |
| # Create a set to remove duplicate packages, and cast it to a list |
| # to keep the return type consistent with get_outdated and |
| # get_uptodate |
| return list({pkg for pkg in packages if pkg.key not in dep_keys}) |
| |
| def iter_packages_latest_infos(self, packages, options): |
| # type: (List[Distribution], Values) -> Iterator[Distribution] |
| with self._build_session(options) as session: |
| finder = self._build_package_finder(options, session) |
| |
| def latest_info(dist): |
| # type: (Distribution) -> Distribution |
| typ = 'unknown' |
| all_candidates = finder.find_all_candidates(dist.key) |
| if not options.pre: |
| # Remove prereleases |
| all_candidates = [candidate for candidate in all_candidates |
| if not candidate.version.is_prerelease] |
| |
| evaluator = finder.make_candidate_evaluator( |
| project_name=dist.project_name, |
| ) |
| best_candidate = evaluator.sort_best_candidate(all_candidates) |
| if best_candidate is None: |
| return None |
| |
| remote_version = best_candidate.version |
| if best_candidate.link.is_wheel: |
| typ = 'wheel' |
| else: |
| typ = 'sdist' |
| # This is dirty but makes the rest of the code much cleaner |
| dist.latest_version = remote_version |
| dist.latest_filetype = typ |
| return dist |
| |
| for dist in map_multithread(latest_info, packages): |
| if dist is not None: |
| yield dist |
| |
| def output_package_listing(self, packages, options): |
| # type: (List[Distribution], Values) -> None |
| packages = sorted( |
| packages, |
| key=lambda dist: dist.project_name.lower(), |
| ) |
| if options.list_format == 'columns' and packages: |
| data, header = format_for_columns(packages, options) |
| self.output_package_listing_columns(data, header) |
| elif options.list_format == 'freeze': |
| for dist in packages: |
| if options.verbose >= 1: |
| write_output("%s==%s (%s)", dist.project_name, |
| dist.version, dist.location) |
| else: |
| write_output("%s==%s", dist.project_name, dist.version) |
| elif options.list_format == 'json': |
| write_output(format_for_json(packages, options)) |
| |
| def output_package_listing_columns(self, data, header): |
| # type: (List[List[str]], List[str]) -> None |
| # insert the header first: we need to know the size of column names |
| if len(data) > 0: |
| data.insert(0, header) |
| |
| pkg_strings, sizes = tabulate(data) |
| |
| # Create and add a separator. |
| if len(data) > 0: |
| pkg_strings.insert(1, " ".join(map(lambda x: '-' * x, sizes))) |
| |
| for val in pkg_strings: |
| write_output(val) |
| |
| |
| def format_for_columns(pkgs, options): |
| # type: (List[Distribution], Values) -> Tuple[List[List[str]], List[str]] |
| """ |
| Convert the package data into something usable |
| by output_package_listing_columns. |
| """ |
| running_outdated = options.outdated |
| # Adjust the header for the `pip list --outdated` case. |
| if running_outdated: |
| header = ["Package", "Version", "Latest", "Type"] |
| else: |
| header = ["Package", "Version"] |
| |
| data = [] |
| if options.verbose >= 1 or any(dist_is_editable(x) for x in pkgs): |
| header.append("Location") |
| if options.verbose >= 1: |
| header.append("Installer") |
| |
| for proj in pkgs: |
| # if we're working on the 'outdated' list, separate out the |
| # latest_version and type |
| row = [proj.project_name, proj.version] |
| |
| if running_outdated: |
| row.append(proj.latest_version) |
| row.append(proj.latest_filetype) |
| |
| if options.verbose >= 1 or dist_is_editable(proj): |
| row.append(proj.location) |
| if options.verbose >= 1: |
| row.append(get_installer(proj)) |
| |
| data.append(row) |
| |
| return data, header |
| |
| |
| def format_for_json(packages, options): |
| # type: (List[Distribution], Values) -> str |
| data = [] |
| for dist in packages: |
| info = { |
| 'name': dist.project_name, |
| 'version': six.text_type(dist.version), |
| } |
| if options.verbose >= 1: |
| info['location'] = dist.location |
| info['installer'] = get_installer(dist) |
| if options.outdated: |
| info['latest_version'] = six.text_type(dist.latest_version) |
| info['latest_filetype'] = dist.latest_filetype |
| data.append(info) |
| return json.dumps(data) |