Spaces:
Running
Running
"""Base option parser setup""" | |
import logging | |
import optparse | |
import shutil | |
import sys | |
import textwrap | |
from contextlib import suppress | |
from typing import Any, Dict, Generator, List, Tuple | |
from pip._internal.cli.status_codes import UNKNOWN_ERROR | |
from pip._internal.configuration import Configuration, ConfigurationError | |
from pip._internal.utils.misc import redact_auth_from_url, strtobool | |
logger = logging.getLogger(__name__) | |
class PrettyHelpFormatter(optparse.IndentedHelpFormatter): | |
"""A prettier/less verbose help formatter for optparse.""" | |
def __init__(self, *args: Any, **kwargs: Any) -> None: | |
# help position must be aligned with __init__.parseopts.description | |
kwargs["max_help_position"] = 30 | |
kwargs["indent_increment"] = 1 | |
kwargs["width"] = shutil.get_terminal_size()[0] - 2 | |
super().__init__(*args, **kwargs) | |
def format_option_strings(self, option: optparse.Option) -> str: | |
return self._format_option_strings(option) | |
def _format_option_strings( | |
self, option: optparse.Option, mvarfmt: str = " <{}>", optsep: str = ", " | |
) -> str: | |
""" | |
Return a comma-separated list of option strings and metavars. | |
:param option: tuple of (short opt, long opt), e.g: ('-f', '--format') | |
:param mvarfmt: metavar format string | |
:param optsep: separator | |
""" | |
opts = [] | |
if option._short_opts: | |
opts.append(option._short_opts[0]) | |
if option._long_opts: | |
opts.append(option._long_opts[0]) | |
if len(opts) > 1: | |
opts.insert(1, optsep) | |
if option.takes_value(): | |
assert option.dest is not None | |
metavar = option.metavar or option.dest.lower() | |
opts.append(mvarfmt.format(metavar.lower())) | |
return "".join(opts) | |
def format_heading(self, heading: str) -> str: | |
if heading == "Options": | |
return "" | |
return heading + ":\n" | |
def format_usage(self, usage: str) -> str: | |
""" | |
Ensure there is only one newline between usage and the first heading | |
if there is no description. | |
""" | |
msg = "\nUsage: {}\n".format(self.indent_lines(textwrap.dedent(usage), " ")) | |
return msg | |
def format_description(self, description: str) -> str: | |
# leave full control over description to us | |
if description: | |
if hasattr(self.parser, "main"): | |
label = "Commands" | |
else: | |
label = "Description" | |
# some doc strings have initial newlines, some don't | |
description = description.lstrip("\n") | |
# some doc strings have final newlines and spaces, some don't | |
description = description.rstrip() | |
# dedent, then reindent | |
description = self.indent_lines(textwrap.dedent(description), " ") | |
description = f"{label}:\n{description}\n" | |
return description | |
else: | |
return "" | |
def format_epilog(self, epilog: str) -> str: | |
# leave full control over epilog to us | |
if epilog: | |
return epilog | |
else: | |
return "" | |
def indent_lines(self, text: str, indent: str) -> str: | |
new_lines = [indent + line for line in text.split("\n")] | |
return "\n".join(new_lines) | |
class UpdatingDefaultsHelpFormatter(PrettyHelpFormatter): | |
"""Custom help formatter for use in ConfigOptionParser. | |
This is updates the defaults before expanding them, allowing | |
them to show up correctly in the help listing. | |
Also redact auth from url type options | |
""" | |
def expand_default(self, option: optparse.Option) -> str: | |
default_values = None | |
if self.parser is not None: | |
assert isinstance(self.parser, ConfigOptionParser) | |
self.parser._update_defaults(self.parser.defaults) | |
assert option.dest is not None | |
default_values = self.parser.defaults.get(option.dest) | |
help_text = super().expand_default(option) | |
if default_values and option.metavar == "URL": | |
if isinstance(default_values, str): | |
default_values = [default_values] | |
# If its not a list, we should abort and just return the help text | |
if not isinstance(default_values, list): | |
default_values = [] | |
for val in default_values: | |
help_text = help_text.replace(val, redact_auth_from_url(val)) | |
return help_text | |
class CustomOptionParser(optparse.OptionParser): | |
def insert_option_group( | |
self, idx: int, *args: Any, **kwargs: Any | |
) -> optparse.OptionGroup: | |
"""Insert an OptionGroup at a given position.""" | |
group = self.add_option_group(*args, **kwargs) | |
self.option_groups.pop() | |
self.option_groups.insert(idx, group) | |
return group | |
def option_list_all(self) -> List[optparse.Option]: | |
"""Get a list of all options, including those in option groups.""" | |
res = self.option_list[:] | |
for i in self.option_groups: | |
res.extend(i.option_list) | |
return res | |
class ConfigOptionParser(CustomOptionParser): | |
"""Custom option parser which updates its defaults by checking the | |
configuration files and environmental variables""" | |
def __init__( | |
self, | |
*args: Any, | |
name: str, | |
isolated: bool = False, | |
**kwargs: Any, | |
) -> None: | |
self.name = name | |
self.config = Configuration(isolated) | |
assert self.name | |
super().__init__(*args, **kwargs) | |
def check_default(self, option: optparse.Option, key: str, val: Any) -> Any: | |
try: | |
return option.check_value(key, val) | |
except optparse.OptionValueError as exc: | |
print(f"An error occurred during configuration: {exc}") | |
sys.exit(3) | |
def _get_ordered_configuration_items( | |
self, | |
) -> Generator[Tuple[str, Any], None, None]: | |
# Configuration gives keys in an unordered manner. Order them. | |
override_order = ["global", self.name, ":env:"] | |
# Pool the options into different groups | |
section_items: Dict[str, List[Tuple[str, Any]]] = { | |
name: [] for name in override_order | |
} | |
for section_key, val in self.config.items(): | |
# ignore empty values | |
if not val: | |
logger.debug( | |
"Ignoring configuration key '%s' as it's value is empty.", | |
section_key, | |
) | |
continue | |
section, key = section_key.split(".", 1) | |
if section in override_order: | |
section_items[section].append((key, val)) | |
# Yield each group in their override order | |
for section in override_order: | |
for key, val in section_items[section]: | |
yield key, val | |
def _update_defaults(self, defaults: Dict[str, Any]) -> Dict[str, Any]: | |
"""Updates the given defaults with values from the config files and | |
the environ. Does a little special handling for certain types of | |
options (lists).""" | |
# Accumulate complex default state. | |
self.values = optparse.Values(self.defaults) | |
late_eval = set() | |
# Then set the options with those values | |
for key, val in self._get_ordered_configuration_items(): | |
# '--' because configuration supports only long names | |
option = self.get_option("--" + key) | |
# Ignore options not present in this parser. E.g. non-globals put | |
# in [global] by users that want them to apply to all applicable | |
# commands. | |
if option is None: | |
continue | |
assert option.dest is not None | |
if option.action in ("store_true", "store_false"): | |
try: | |
val = strtobool(val) | |
except ValueError: | |
self.error( | |
"{} is not a valid value for {} option, " # noqa | |
"please specify a boolean value like yes/no, " | |
"true/false or 1/0 instead.".format(val, key) | |
) | |
elif option.action == "count": | |
with suppress(ValueError): | |
val = strtobool(val) | |
with suppress(ValueError): | |
val = int(val) | |
if not isinstance(val, int) or val < 0: | |
self.error( | |
"{} is not a valid value for {} option, " # noqa | |
"please instead specify either a non-negative integer " | |
"or a boolean value like yes/no or false/true " | |
"which is equivalent to 1/0.".format(val, key) | |
) | |
elif option.action == "append": | |
val = val.split() | |
val = [self.check_default(option, key, v) for v in val] | |
elif option.action == "callback": | |
assert option.callback is not None | |
late_eval.add(option.dest) | |
opt_str = option.get_opt_string() | |
val = option.convert_value(opt_str, val) | |
# From take_action | |
args = option.callback_args or () | |
kwargs = option.callback_kwargs or {} | |
option.callback(option, opt_str, val, self, *args, **kwargs) | |
else: | |
val = self.check_default(option, key, val) | |
defaults[option.dest] = val | |
for key in late_eval: | |
defaults[key] = getattr(self.values, key) | |
self.values = None | |
return defaults | |
def get_default_values(self) -> optparse.Values: | |
"""Overriding to make updating the defaults after instantiation of | |
the option parser possible, _update_defaults() does the dirty work.""" | |
if not self.process_default_values: | |
# Old, pre-Optik 1.5 behaviour. | |
return optparse.Values(self.defaults) | |
# Load the configuration, or error out in case of an error | |
try: | |
self.config.load() | |
except ConfigurationError as err: | |
self.exit(UNKNOWN_ERROR, str(err)) | |
defaults = self._update_defaults(self.defaults.copy()) # ours | |
for option in self._get_all_options(): | |
assert option.dest is not None | |
default = defaults.get(option.dest) | |
if isinstance(default, str): | |
opt_str = option.get_opt_string() | |
defaults[option.dest] = option.check_value(opt_str, default) | |
return optparse.Values(defaults) | |
def error(self, msg: str) -> None: | |
self.print_usage(sys.stderr) | |
self.exit(UNKNOWN_ERROR, f"{msg}\n") | |