Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/pip/_internal/commands/search.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 import logging | |
2 import shutil | |
3 import sys | |
4 import textwrap | |
5 from collections import OrderedDict | |
6 | |
7 from pip._vendor import pkg_resources | |
8 from pip._vendor.packaging.version import parse as parse_version | |
9 | |
10 # NOTE: XMLRPC Client is not annotated in typeshed as on 2017-07-17, which is | |
11 # why we ignore the type on this import | |
12 from pip._vendor.six.moves import xmlrpc_client # type: ignore | |
13 | |
14 from pip._internal.cli.base_command import Command | |
15 from pip._internal.cli.req_command import SessionCommandMixin | |
16 from pip._internal.cli.status_codes import NO_MATCHES_FOUND, SUCCESS | |
17 from pip._internal.exceptions import CommandError | |
18 from pip._internal.models.index import PyPI | |
19 from pip._internal.network.xmlrpc import PipXmlrpcTransport | |
20 from pip._internal.utils.logging import indent_log | |
21 from pip._internal.utils.misc import get_distribution, write_output | |
22 from pip._internal.utils.typing import MYPY_CHECK_RUNNING | |
23 | |
24 if MYPY_CHECK_RUNNING: | |
25 from optparse import Values | |
26 from typing import Dict, List, Optional | |
27 | |
28 from typing_extensions import TypedDict | |
29 TransformedHit = TypedDict( | |
30 'TransformedHit', | |
31 {'name': str, 'summary': str, 'versions': List[str]}, | |
32 ) | |
33 | |
34 logger = logging.getLogger(__name__) | |
35 | |
36 | |
37 class SearchCommand(Command, SessionCommandMixin): | |
38 """Search for PyPI packages whose name or summary contains <query>.""" | |
39 | |
40 usage = """ | |
41 %prog [options] <query>""" | |
42 ignore_require_venv = True | |
43 | |
44 def add_options(self): | |
45 # type: () -> None | |
46 self.cmd_opts.add_option( | |
47 '-i', '--index', | |
48 dest='index', | |
49 metavar='URL', | |
50 default=PyPI.pypi_url, | |
51 help='Base URL of Python Package Index (default %default)') | |
52 | |
53 self.parser.insert_option_group(0, self.cmd_opts) | |
54 | |
55 def run(self, options, args): | |
56 # type: (Values, List[str]) -> int | |
57 if not args: | |
58 raise CommandError('Missing required argument (search query).') | |
59 query = args | |
60 pypi_hits = self.search(query, options) | |
61 hits = transform_hits(pypi_hits) | |
62 | |
63 terminal_width = None | |
64 if sys.stdout.isatty(): | |
65 terminal_width = shutil.get_terminal_size()[0] | |
66 | |
67 print_results(hits, terminal_width=terminal_width) | |
68 if pypi_hits: | |
69 return SUCCESS | |
70 return NO_MATCHES_FOUND | |
71 | |
72 def search(self, query, options): | |
73 # type: (List[str], Values) -> List[Dict[str, str]] | |
74 index_url = options.index | |
75 | |
76 session = self.get_default_session(options) | |
77 | |
78 transport = PipXmlrpcTransport(index_url, session) | |
79 pypi = xmlrpc_client.ServerProxy(index_url, transport) | |
80 try: | |
81 hits = pypi.search({'name': query, 'summary': query}, 'or') | |
82 except xmlrpc_client.Fault as fault: | |
83 message = "XMLRPC request failed [code: {code}]\n{string}".format( | |
84 code=fault.faultCode, | |
85 string=fault.faultString, | |
86 ) | |
87 raise CommandError(message) | |
88 return hits | |
89 | |
90 | |
91 def transform_hits(hits): | |
92 # type: (List[Dict[str, str]]) -> List[TransformedHit] | |
93 """ | |
94 The list from pypi is really a list of versions. We want a list of | |
95 packages with the list of versions stored inline. This converts the | |
96 list from pypi into one we can use. | |
97 """ | |
98 packages = OrderedDict() # type: OrderedDict[str, TransformedHit] | |
99 for hit in hits: | |
100 name = hit['name'] | |
101 summary = hit['summary'] | |
102 version = hit['version'] | |
103 | |
104 if name not in packages.keys(): | |
105 packages[name] = { | |
106 'name': name, | |
107 'summary': summary, | |
108 'versions': [version], | |
109 } | |
110 else: | |
111 packages[name]['versions'].append(version) | |
112 | |
113 # if this is the highest version, replace summary and score | |
114 if version == highest_version(packages[name]['versions']): | |
115 packages[name]['summary'] = summary | |
116 | |
117 return list(packages.values()) | |
118 | |
119 | |
120 def print_results(hits, name_column_width=None, terminal_width=None): | |
121 # type: (List[TransformedHit], Optional[int], Optional[int]) -> None | |
122 if not hits: | |
123 return | |
124 if name_column_width is None: | |
125 name_column_width = max([ | |
126 len(hit['name']) + len(highest_version(hit.get('versions', ['-']))) | |
127 for hit in hits | |
128 ]) + 4 | |
129 | |
130 installed_packages = [p.project_name for p in pkg_resources.working_set] | |
131 for hit in hits: | |
132 name = hit['name'] | |
133 summary = hit['summary'] or '' | |
134 latest = highest_version(hit.get('versions', ['-'])) | |
135 if terminal_width is not None: | |
136 target_width = terminal_width - name_column_width - 5 | |
137 if target_width > 10: | |
138 # wrap and indent summary to fit terminal | |
139 summary_lines = textwrap.wrap(summary, target_width) | |
140 summary = ('\n' + ' ' * (name_column_width + 3)).join( | |
141 summary_lines) | |
142 | |
143 line = '{name_latest:{name_column_width}} - {summary}'.format( | |
144 name_latest='{name} ({latest})'.format(**locals()), | |
145 **locals()) | |
146 try: | |
147 write_output(line) | |
148 if name in installed_packages: | |
149 dist = get_distribution(name) | |
150 assert dist is not None | |
151 with indent_log(): | |
152 if dist.version == latest: | |
153 write_output('INSTALLED: %s (latest)', dist.version) | |
154 else: | |
155 write_output('INSTALLED: %s', dist.version) | |
156 if parse_version(latest).pre: | |
157 write_output('LATEST: %s (pre-release; install' | |
158 ' with "pip install --pre")', latest) | |
159 else: | |
160 write_output('LATEST: %s', latest) | |
161 except UnicodeEncodeError: | |
162 pass | |
163 | |
164 | |
165 def highest_version(versions): | |
166 # type: (List[str]) -> str | |
167 return max(versions, key=parse_version) |