140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
import logging
|
|
from optparse import Values
|
|
from typing import Any, Iterable, List, Optional
|
|
|
|
from pip._vendor.packaging.version import Version
|
|
|
|
from pip._internal.cli import cmdoptions
|
|
from pip._internal.cli.req_command import IndexGroupCommand
|
|
from pip._internal.cli.status_codes import ERROR, SUCCESS
|
|
from pip._internal.commands.search import print_dist_installation_info
|
|
from pip._internal.exceptions import CommandError, DistributionNotFound, PipError
|
|
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.models.target_python import TargetPython
|
|
from pip._internal.network.session import PipSession
|
|
from pip._internal.utils.misc import write_output
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class IndexCommand(IndexGroupCommand):
|
|
"""
|
|
Inspect information available from package indexes.
|
|
"""
|
|
|
|
ignore_require_venv = True
|
|
usage = """
|
|
%prog versions <package>
|
|
"""
|
|
|
|
def add_options(self) -> None:
|
|
cmdoptions.add_target_python_options(self.cmd_opts)
|
|
|
|
self.cmd_opts.add_option(cmdoptions.ignore_requires_python())
|
|
self.cmd_opts.add_option(cmdoptions.pre())
|
|
self.cmd_opts.add_option(cmdoptions.no_binary())
|
|
self.cmd_opts.add_option(cmdoptions.only_binary())
|
|
|
|
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 run(self, options: Values, args: List[str]) -> int:
|
|
handlers = {
|
|
"versions": self.get_available_package_versions,
|
|
}
|
|
|
|
logger.warning(
|
|
"pip index is currently an experimental command. "
|
|
"It may be removed/changed in a future release "
|
|
"without prior warning."
|
|
)
|
|
|
|
# Determine action
|
|
if not args or args[0] not in handlers:
|
|
logger.error(
|
|
"Need an action (%s) to perform.",
|
|
", ".join(sorted(handlers)),
|
|
)
|
|
return ERROR
|
|
|
|
action = args[0]
|
|
|
|
# Error handling happens here, not in the action-handlers.
|
|
try:
|
|
handlers[action](options, args[1:])
|
|
except PipError as e:
|
|
logger.error(e.args[0])
|
|
return ERROR
|
|
|
|
return SUCCESS
|
|
|
|
def _build_package_finder(
|
|
self,
|
|
options: Values,
|
|
session: PipSession,
|
|
target_python: Optional[TargetPython] = None,
|
|
ignore_requires_python: Optional[bool] = None,
|
|
) -> PackageFinder:
|
|
"""
|
|
Create a package finder appropriate to the index 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,
|
|
ignore_requires_python=ignore_requires_python,
|
|
)
|
|
|
|
return PackageFinder.create(
|
|
link_collector=link_collector,
|
|
selection_prefs=selection_prefs,
|
|
target_python=target_python,
|
|
)
|
|
|
|
def get_available_package_versions(self, options: Values, args: List[Any]) -> None:
|
|
if len(args) != 1:
|
|
raise CommandError("You need to specify exactly one argument")
|
|
|
|
target_python = cmdoptions.make_target_python(options)
|
|
query = args[0]
|
|
|
|
with self._build_session(options) as session:
|
|
finder = self._build_package_finder(
|
|
options=options,
|
|
session=session,
|
|
target_python=target_python,
|
|
ignore_requires_python=options.ignore_requires_python,
|
|
)
|
|
|
|
versions: Iterable[Version] = (
|
|
candidate.version for candidate in finder.find_all_candidates(query)
|
|
)
|
|
|
|
if not options.pre:
|
|
# Remove prereleases
|
|
versions = (
|
|
version for version in versions if not version.is_prerelease
|
|
)
|
|
versions = set(versions)
|
|
|
|
if not versions:
|
|
raise DistributionNotFound(
|
|
f"No matching distribution found for {query}"
|
|
)
|
|
|
|
formatted_versions = [str(ver) for ver in sorted(versions, reverse=True)]
|
|
latest = formatted_versions[0]
|
|
|
|
write_output(f"{query} ({latest})")
|
|
write_output("Available versions: {}".format(", ".join(formatted_versions)))
|
|
print_dist_installation_info(query, latest)
|