python (3.11.7)
1 """Base Command class, and related routines"""
2
3 import functools
4 import logging
5 import logging.config
6 import optparse
7 import os
8 import sys
9 import traceback
10 from optparse import Values
11 from typing import Any, Callable, List, Optional, Tuple
12
13 from pip._vendor.rich import traceback as rich_traceback
14
15 from pip._internal.cli import cmdoptions
16 from pip._internal.cli.command_context import CommandContextMixIn
17 from pip._internal.cli.parser import ConfigOptionParser, UpdatingDefaultsHelpFormatter
18 from pip._internal.cli.status_codes import (
19 ERROR,
20 PREVIOUS_BUILD_DIR_ERROR,
21 UNKNOWN_ERROR,
22 VIRTUALENV_NOT_FOUND,
23 )
24 from pip._internal.exceptions import (
25 BadCommand,
26 CommandError,
27 DiagnosticPipError,
28 InstallationError,
29 NetworkConnectionError,
30 PreviousBuildDirError,
31 UninstallationError,
32 )
33 from pip._internal.utils.filesystem import check_path_owner
34 from pip._internal.utils.logging import BrokenStdoutLoggingError, setup_logging
35 from pip._internal.utils.misc import get_prog, normalize_path
36 from pip._internal.utils.temp_dir import TempDirectoryTypeRegistry as TempDirRegistry
37 from pip._internal.utils.temp_dir import global_tempdir_manager, tempdir_registry
38 from pip._internal.utils.virtualenv import running_under_virtualenv
39
40 __all__ = ["Command"]
41
42 logger = logging.getLogger(__name__)
43
44
45 class ESC[4;38;5;81mCommand(ESC[4;38;5;149mCommandContextMixIn):
46 usage: str = ""
47 ignore_require_venv: bool = False
48
49 def __init__(self, name: str, summary: str, isolated: bool = False) -> None:
50 super().__init__()
51
52 self.name = name
53 self.summary = summary
54 self.parser = ConfigOptionParser(
55 usage=self.usage,
56 prog=f"{get_prog()} {name}",
57 formatter=UpdatingDefaultsHelpFormatter(),
58 add_help_option=False,
59 name=name,
60 description=self.__doc__,
61 isolated=isolated,
62 )
63
64 self.tempdir_registry: Optional[TempDirRegistry] = None
65
66 # Commands should add options to this option group
67 optgroup_name = f"{self.name.capitalize()} Options"
68 self.cmd_opts = optparse.OptionGroup(self.parser, optgroup_name)
69
70 # Add the general options
71 gen_opts = cmdoptions.make_option_group(
72 cmdoptions.general_group,
73 self.parser,
74 )
75 self.parser.add_option_group(gen_opts)
76
77 self.add_options()
78
79 def add_options(self) -> None:
80 pass
81
82 def handle_pip_version_check(self, options: Values) -> None:
83 """
84 This is a no-op so that commands by default do not do the pip version
85 check.
86 """
87 # Make sure we do the pip version check if the index_group options
88 # are present.
89 assert not hasattr(options, "no_index")
90
91 def run(self, options: Values, args: List[str]) -> int:
92 raise NotImplementedError
93
94 def parse_args(self, args: List[str]) -> Tuple[Values, List[str]]:
95 # factored out for testability
96 return self.parser.parse_args(args)
97
98 def main(self, args: List[str]) -> int:
99 try:
100 with self.main_context():
101 return self._main(args)
102 finally:
103 logging.shutdown()
104
105 def _main(self, args: List[str]) -> int:
106 # We must initialize this before the tempdir manager, otherwise the
107 # configuration would not be accessible by the time we clean up the
108 # tempdir manager.
109 self.tempdir_registry = self.enter_context(tempdir_registry())
110 # Intentionally set as early as possible so globally-managed temporary
111 # directories are available to the rest of the code.
112 self.enter_context(global_tempdir_manager())
113
114 options, args = self.parse_args(args)
115
116 # Set verbosity so that it can be used elsewhere.
117 self.verbosity = options.verbose - options.quiet
118
119 level_number = setup_logging(
120 verbosity=self.verbosity,
121 no_color=options.no_color,
122 user_log_file=options.log,
123 )
124
125 always_enabled_features = set(options.features_enabled) & set(
126 cmdoptions.ALWAYS_ENABLED_FEATURES
127 )
128 if always_enabled_features:
129 logger.warning(
130 "The following features are always enabled: %s. ",
131 ", ".join(sorted(always_enabled_features)),
132 )
133
134 # Make sure that the --python argument isn't specified after the
135 # subcommand. We can tell, because if --python was specified,
136 # we should only reach this point if we're running in the created
137 # subprocess, which has the _PIP_RUNNING_IN_SUBPROCESS environment
138 # variable set.
139 if options.python and "_PIP_RUNNING_IN_SUBPROCESS" not in os.environ:
140 logger.critical(
141 "The --python option must be placed before the pip subcommand name"
142 )
143 sys.exit(ERROR)
144
145 # TODO: Try to get these passing down from the command?
146 # without resorting to os.environ to hold these.
147 # This also affects isolated builds and it should.
148
149 if options.no_input:
150 os.environ["PIP_NO_INPUT"] = "1"
151
152 if options.exists_action:
153 os.environ["PIP_EXISTS_ACTION"] = " ".join(options.exists_action)
154
155 if options.require_venv and not self.ignore_require_venv:
156 # If a venv is required check if it can really be found
157 if not running_under_virtualenv():
158 logger.critical("Could not find an activated virtualenv (required).")
159 sys.exit(VIRTUALENV_NOT_FOUND)
160
161 if options.cache_dir:
162 options.cache_dir = normalize_path(options.cache_dir)
163 if not check_path_owner(options.cache_dir):
164 logger.warning(
165 "The directory '%s' or its parent directory is not owned "
166 "or is not writable by the current user. The cache "
167 "has been disabled. Check the permissions and owner of "
168 "that directory. If executing pip with sudo, you should "
169 "use sudo's -H flag.",
170 options.cache_dir,
171 )
172 options.cache_dir = None
173
174 def intercepts_unhandled_exc(
175 run_func: Callable[..., int]
176 ) -> Callable[..., int]:
177 @functools.wraps(run_func)
178 def exc_logging_wrapper(*args: Any) -> int:
179 try:
180 status = run_func(*args)
181 assert isinstance(status, int)
182 return status
183 except DiagnosticPipError as exc:
184 logger.error("[present-rich] %s", exc)
185 logger.debug("Exception information:", exc_info=True)
186
187 return ERROR
188 except PreviousBuildDirError as exc:
189 logger.critical(str(exc))
190 logger.debug("Exception information:", exc_info=True)
191
192 return PREVIOUS_BUILD_DIR_ERROR
193 except (
194 InstallationError,
195 UninstallationError,
196 BadCommand,
197 NetworkConnectionError,
198 ) as exc:
199 logger.critical(str(exc))
200 logger.debug("Exception information:", exc_info=True)
201
202 return ERROR
203 except CommandError as exc:
204 logger.critical("%s", exc)
205 logger.debug("Exception information:", exc_info=True)
206
207 return ERROR
208 except BrokenStdoutLoggingError:
209 # Bypass our logger and write any remaining messages to
210 # stderr because stdout no longer works.
211 print("ERROR: Pipe to stdout was broken", file=sys.stderr)
212 if level_number <= logging.DEBUG:
213 traceback.print_exc(file=sys.stderr)
214
215 return ERROR
216 except KeyboardInterrupt:
217 logger.critical("Operation cancelled by user")
218 logger.debug("Exception information:", exc_info=True)
219
220 return ERROR
221 except BaseException:
222 logger.critical("Exception:", exc_info=True)
223
224 return UNKNOWN_ERROR
225
226 return exc_logging_wrapper
227
228 try:
229 if not options.debug_mode:
230 run = intercepts_unhandled_exc(self.run)
231 else:
232 run = self.run
233 rich_traceback.install(show_locals=True)
234 return run(options, args)
235 finally:
236 self.handle_pip_version_check(options)
Browse
Usage
Versions