reporter.py 3.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  1. from __future__ import annotations
  2. from collections import defaultdict
  3. from collections.abc import Mapping
  4. from logging import getLogger
  5. from typing import Any
  6. from pip._vendor.resolvelib.reporters import BaseReporter
  7. from .base import Candidate, Constraint, Requirement
  8. logger = getLogger(__name__)
  9. class PipReporter(BaseReporter[Requirement, Candidate, str]):
  10. def __init__(self, constraints: Mapping[str, Constraint] | None = None) -> None:
  11. self.reject_count_by_package: defaultdict[str, int] = defaultdict(int)
  12. self._constraints = constraints or {}
  13. self._messages_at_reject_count = {
  14. 1: (
  15. "pip is looking at multiple versions of {package_name} to "
  16. "determine which version is compatible with other "
  17. "requirements. This could take a while."
  18. ),
  19. 8: (
  20. "pip is still looking at multiple versions of {package_name} to "
  21. "determine which version is compatible with other "
  22. "requirements. This could take a while."
  23. ),
  24. 13: (
  25. "This is taking longer than usual. You might need to provide "
  26. "the dependency resolver with stricter constraints to reduce "
  27. "runtime. See https://pip.pypa.io/warnings/backtracking for "
  28. "guidance. If you want to abort this run, press Ctrl + C."
  29. ),
  30. }
  31. def rejecting_candidate(self, criterion: Any, candidate: Candidate) -> None:
  32. """Report a candidate being rejected.
  33. Logs both the rejection count message (if applicable) and details about
  34. the requirements and constraints that caused the rejection.
  35. """
  36. self.reject_count_by_package[candidate.name] += 1
  37. count = self.reject_count_by_package[candidate.name]
  38. if count in self._messages_at_reject_count:
  39. message = self._messages_at_reject_count[count]
  40. logger.info("INFO: %s", message.format(package_name=candidate.name))
  41. msg = "Will try a different candidate, due to conflict:"
  42. for req_info in criterion.information:
  43. req, parent = req_info.requirement, req_info.parent
  44. msg += "\n "
  45. if parent:
  46. msg += f"{parent.name} {parent.version} depends on "
  47. else:
  48. msg += "The user requested "
  49. msg += req.format_for_error()
  50. # Add any relevant constraints
  51. if self._constraints:
  52. name = candidate.name
  53. constraint = self._constraints.get(name)
  54. if constraint and constraint.specifier:
  55. constraint_text = f"{name}{constraint.specifier}"
  56. msg += f"\n The user requested (constraint) {constraint_text}"
  57. logger.debug(msg)
  58. class PipDebuggingReporter(BaseReporter[Requirement, Candidate, str]):
  59. """A reporter that does an info log for every event it sees."""
  60. def starting(self) -> None:
  61. logger.info("Reporter.starting()")
  62. def starting_round(self, index: int) -> None:
  63. logger.info("Reporter.starting_round(%r)", index)
  64. def ending_round(self, index: int, state: Any) -> None:
  65. logger.info("Reporter.ending_round(%r, state)", index)
  66. logger.debug("Reporter.ending_round(%r, %r)", index, state)
  67. def ending(self, state: Any) -> None:
  68. logger.info("Reporter.ending(%r)", state)
  69. def adding_requirement(
  70. self, requirement: Requirement, parent: Candidate | None
  71. ) -> None:
  72. logger.info("Reporter.adding_requirement(%r, %r)", requirement, parent)
  73. def rejecting_candidate(self, criterion: Any, candidate: Candidate) -> None:
  74. logger.info("Reporter.rejecting_candidate(%r, %r)", criterion, candidate)
  75. def pinning(self, candidate: Candidate) -> None:
  76. logger.info("Reporter.pinning(%r)", candidate)