140 lines
		
	
	
	
		
			4.7 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
		
		
			
		
	
	
			140 lines
		
	
	
	
		
			4.7 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
|   | import logging | ||
|  | from optparse import Values | ||
|  | from typing import Any, Iterable, List, Optional, Union | ||
|  | 
 | ||
|  | from pip._vendor.packaging.version import LegacyVersion, 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[Union[LegacyVersion, 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) |