Skip to content

Commit

Permalink
Rework how --debug is handled in main
Browse files Browse the repository at this point in the history
Move the `run` call into a wrapper function and inline the exception
handling while moving the try-finally around `run` into the
nested function.

This reduces the amount of code at a deeper nesting level and moves the
self-version-check logic into the "around the run" exception catching so
that a consistent message is presented in cases of failures.
  • Loading branch information
pradyunsg committed Jul 21, 2024
1 parent be21d82 commit 3518d32
Showing 1 changed file with 59 additions and 64 deletions.
123 changes: 59 additions & 64 deletions src/pip/_internal/cli/base_command.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,13 @@
"""Base Command class, and related routines"""

import functools
import logging
import logging.config
import optparse
import os
import sys
import traceback
from optparse import Values
from typing import Any, Callable, List, Optional, Tuple
from typing import List, Optional, Tuple

from pip._vendor.rich import reconfigure
from pip._vendor.rich import traceback as rich_traceback
Expand Down Expand Up @@ -91,6 +90,63 @@ def handle_pip_version_check(self, options: Values) -> None:
def run(self, options: Values, args: List[str]) -> int:
raise NotImplementedError

def _run_wrapper(self, level_number: int, options: Values, args: List[str]) -> int:
def _inner_run() -> int:
try:
return self.run(options, args)
finally:
self.handle_pip_version_check(options)

if options.debug_mode:
rich_traceback.install(show_locals=True)
return _inner_run()

try:
status = _inner_run()
assert isinstance(status, int)
return status
except DiagnosticPipError as exc:
logger.error("%s", exc, extra={"rich": True})
logger.debug("Exception information:", exc_info=True)

return ERROR
except PreviousBuildDirError as exc:
logger.critical(str(exc))
logger.debug("Exception information:", exc_info=True)

return PREVIOUS_BUILD_DIR_ERROR
except (
InstallationError,
BadCommand,
NetworkConnectionError,
) as exc:
logger.critical(str(exc))
logger.debug("Exception information:", exc_info=True)

return ERROR
except CommandError as exc:
logger.critical("%s", exc)
logger.debug("Exception information:", exc_info=True)

return ERROR
except BrokenStdoutLoggingError:
# Bypass our logger and write any remaining messages to
# stderr because stdout no longer works.
print("ERROR: Pipe to stdout was broken", file=sys.stderr)
if level_number <= logging.DEBUG:
traceback.print_exc(file=sys.stderr)

return ERROR
except KeyboardInterrupt:
logger.critical("Operation cancelled by user")
logger.debug("Exception information:", exc_info=True)

return ERROR
except BaseException:
logger.critical("Exception:", exc_info=True)

return UNKNOWN_ERROR

def parse_args(self, args: List[str]) -> Tuple[Values, List[str]]:
# factored out for testability
return self.parser.parse_args(args)
Expand Down Expand Up @@ -172,65 +228,4 @@ def _main(self, args: List[str]) -> int:
)
options.cache_dir = None

def intercepts_unhandled_exc(
run_func: Callable[..., int]
) -> Callable[..., int]:
@functools.wraps(run_func)
def exc_logging_wrapper(*args: Any) -> int:
try:
status = run_func(*args)
assert isinstance(status, int)
return status
except DiagnosticPipError as exc:
logger.error("%s", exc, extra={"rich": True})
logger.debug("Exception information:", exc_info=True)

return ERROR
except PreviousBuildDirError as exc:
logger.critical(str(exc))
logger.debug("Exception information:", exc_info=True)

return PREVIOUS_BUILD_DIR_ERROR
except (
InstallationError,
BadCommand,
NetworkConnectionError,
) as exc:
logger.critical(str(exc))
logger.debug("Exception information:", exc_info=True)

return ERROR
except CommandError as exc:
logger.critical("%s", exc)
logger.debug("Exception information:", exc_info=True)

return ERROR
except BrokenStdoutLoggingError:
# Bypass our logger and write any remaining messages to
# stderr because stdout no longer works.
print("ERROR: Pipe to stdout was broken", file=sys.stderr)
if level_number <= logging.DEBUG:
traceback.print_exc(file=sys.stderr)

return ERROR
except KeyboardInterrupt:
logger.critical("Operation cancelled by user")
logger.debug("Exception information:", exc_info=True)

return ERROR
except BaseException:
logger.critical("Exception:", exc_info=True)

return UNKNOWN_ERROR

return exc_logging_wrapper

try:
if not options.debug_mode:
run = intercepts_unhandled_exc(self.run)
else:
run = self.run
rich_traceback.install(show_locals=True)
return run(options, args)
finally:
self.handle_pip_version_check(options)
return self._run_wrapper(level_number, options, args)

0 comments on commit 3518d32

Please sign in to comment.