Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Improve logging #11

Merged
merged 6 commits into from
Jun 10, 2024
Merged

Improve logging #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
4a57dbb
Improve logging
realshouzy Jun 10, 2024
d0bce10
Futher improve logging
realshouzy Jun 10, 2024
f062889
Adjust tests for new logging
realshouzy Jun 10, 2024
6553516
Fix testing fails
realshouzy Jun 10, 2024
fea92e4
Mark `retain_pytest_handlers` as in internal
realshouzy Jun 10, 2024
6635adc
Update pragam
realshouzy Jun 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 61 additions & 21 deletions pip_manage/_logging.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,41 +1,81 @@
from __future__ import annotations

__all__: list[str] = ["setup_logging", "set_logging_level"]
__all__: list[str] = ["setup_logging"]

import logging
import logging.config
import sys
from typing import TextIO
from typing import ClassVar, Literal

if sys.version_info >= (3, 12): # pragma: >=3.12 cover
from typing import override
else: # pragma: <3.12 cover
from typing_extensions import override


class _StdOutFilter(logging.Filter):
class _NonErrorFilter(logging.Filter):
@override
def filter(self, record: logging.LogRecord) -> bool:
return record.levelno <= logging.INFO


def setup_logging(logger_name: str) -> logging.Logger:
logger: logging.Logger = logging.getLogger(logger_name)
class _ColoredFormatter(logging.Formatter):
COLORS: ClassVar[dict[str, str]] = {
"DEBUG": "\x1b[0;37m",
"INFO": "\x1b[0;32m",
"WARNING": "\x1b[0;33m",
"ERROR": "\x1b[0;31m",
"CRITICAL": "\x1b[1;31m",
}
RESET: ClassVar[Literal["\x1b[0m"]] = "\x1b[0m"

stdout_handler: logging.StreamHandler[TextIO] = logging.StreamHandler(sys.stdout)
stdout_handler.set_name("stdout")
stdout_handler.addFilter(_StdOutFilter())
stdout_handler.setFormatter(logging.Formatter("%(message)s"))
stdout_handler.setLevel(logging.DEBUG)

stderr_handler: logging.StreamHandler[TextIO] = logging.StreamHandler(sys.stderr)
stderr_handler.set_name("stderr")
stderr_handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
stderr_handler.setLevel(logging.WARNING)

logger.addHandler(stderr_handler)
logger.addHandler(stdout_handler)
return logger
@override
def format(self, record: logging.LogRecord) -> str:
log_color: str = self.COLORS.get(record.levelname, self.RESET)
record.msg = f"{log_color}{record.levelname}: {record.msg}{self.RESET}"
return super().format(record)


def set_logging_level(logger: logging.Logger, *, verbose: bool) -> None:
logger.setLevel(logging.DEBUG if verbose else logging.INFO)
def setup_logging(logger_name: str, *, verbose: bool) -> None:
level: Literal["DEBUG", "INFO"] = "DEBUG" if verbose else "INFO"
logging.config.dictConfig(
{
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"simple": {
"format": "%(message)s",
},
"colored": {
"()": _ColoredFormatter,
},
},
"filters": {
"StdOutFilter": {
"()": _NonErrorFilter,
},
},
"handlers": {
"stdout": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stdout",
"formatter": "simple",
"filters": ["StdOutFilter"],
"level": "DEBUG",
},
"stderr": {
"class": "logging.StreamHandler",
"stream": "ext://sys.stderr",
"formatter": "colored",
"level": "WARNING",
},
},
"loggers": {
"": {
"level": "DEBUG",
"handlers": ["stdout", "stderr"],
},
logger_name: {"level": level, "propagate": True},
},
},
)
62 changes: 36 additions & 26 deletions pip_manage/pip_purge.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,11 @@

import argparse
import importlib.metadata
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Final, NamedTuple

from pip_manage._logging import set_logging_level, setup_logging
from pip_manage._logging import setup_logging
from pip_manage._pip_interface import (
PIP_CMD,
UNINSTALL_ONLY,
Expand All @@ -18,7 +19,6 @@
)

if TYPE_CHECKING:
import logging
from collections.abc import Sequence

_EPILOG: Final[str] = (
Expand All @@ -29,8 +29,6 @@
"""
)

_logger: logging.Logger = setup_logging(__title__)


def _parse_args(
args: Sequence[str] | None = None,
Expand Down Expand Up @@ -189,42 +187,50 @@ def main( # pylint: disable=R0914, R0915 # noqa: PLR0915
) -> int:
args, forwarded = _parse_args(argv)
uninstall_args: list[str] = filter_forwards_include(forwarded, UNINSTALL_ONLY)
set_logging_level(_logger, verbose=args.verbose)
_logger.debug("Forwarded arguments: %s", forwarded)
_logger.debug("Arguments forwarded to 'pip uninstall': %s", uninstall_args)
setup_logging(__title__, verbose=args.verbose)
logger: logging.Logger = logging.getLogger(__title__)

logger.debug("Forwarded arguments: %s", forwarded)
logger.debug("Arguments forwarded to 'pip uninstall': %s", uninstall_args)

if unrecognized_args := set(forwarded).difference(uninstall_args):
formatted_unrecognized_arg: list[str] = [
f"'{unrecognized_arg}'" for unrecognized_arg in sorted(unrecognized_args)
]
_logger.warning(
logger.warning(
"Unrecognized arguments: %s",
", ".join(formatted_unrecognized_arg),
)
try:
requirements: list[str] = _read_from_requirements(args.requirements)
except OSError as err:
logger.error("Could not open requirements file: %s", err)
return 1

if not (packages := [*args.packages, *_read_from_requirements(args.requirements)]):
_logger.error("No packages provided")
if not (packages := [*args.packages, *requirements]):
logger.error("You must give at least one requirement to uninstall")
return 1

package_dependencies: dict[str, _DependencyInfo] = {}
for package in packages:
if not _is_installed(package):
_logger.warning("%s is not installed", package)
logger.warning("Skipping %s as it is not installed", package)
continue

if package in args.exclude:
logger.debug("Skipping %s", package)
continue

package_dependencies[package] = dependency_info = _get_dependencies_of_package(
package,
ignore_extra=args.ignore_extra,
)
_logger.debug(
logger.debug(
"%s requires: %s",
package,
dependency_info.dependencies,
)
_logger.debug(
logger.debug(
"%s is required by: %s",
package,
dependency_info.dependents,
Expand All @@ -238,12 +244,12 @@ def main( # pylint: disable=R0914, R0915 # noqa: PLR0915
dependent_package,
ignore_extra=args.ignore_extra,
)
_logger.debug(
logger.debug(
"%s requires: %s",
dependent_package,
dependent_package_dependency_info.dependencies,
)
_logger.debug(
logger.debug(
"%s is required by: %s",
dependent_package,
dependent_package_dependency_info.dependents,
Expand All @@ -254,11 +260,11 @@ def main( # pylint: disable=R0914, R0915 # noqa: PLR0915
# If a package has dependents that are NOT supposed to also by uninstalled,
# it removes the package from package_dependencies.
for package_name, dependency_info in package_dependencies.copy().items():
_logger.debug("Checking %s", package_name)
logger.debug("Checking %s", package_name)
if dependency_info.dependents and not all(
package in package_dependencies for package in dependency_info.dependents
):
_logger.info(
logger.info(
"Cannot uninstall %s, required by: %s",
package_name,
", ".join(dependency_info.dependents.difference(package_dependencies)),
Expand All @@ -270,34 +276,38 @@ def main( # pylint: disable=R0914, R0915 # noqa: PLR0915
# are also reconsidered.
packages_to_uninstall: list[str] = []
for package_name, dependency_info in package_dependencies.items():
_logger.debug("Checking %s again", package_name)
logger.debug("Checking %s again", package_name)
if not dependency_info.dependents or all(
package in package_dependencies for package in dependency_info.dependents
):
packages_to_uninstall.append(package_name)
_logger.debug("%s will be uninstalled", package_name)
logger.debug("%s will be uninstalled", package_name)
else:
_logger.info(
logger.info(
"Cannot uninstall %s, required by: %s",
package_name,
", ".join(dependency_info.dependents.difference(package_dependencies)),
)

_logger.debug("Finished checking packages")
logger.debug("Finished checking packages")

if not packages_to_uninstall:
_logger.info("No packages to purge")
logger.info("No packages to purge")
return 0

packages_to_uninstall.sort()
_logger.info(
logger.info(
"The following packages will be uninstalled: %s",
", ".join(packages_to_uninstall),
)

if args.freeze_purged_packages:
_freeze_packages(args.freeze_file, packages_to_uninstall)
_logger.debug("Wrote packages to %s", args.freeze_file)
try:
_freeze_packages(args.freeze_file, packages_to_uninstall)
except OSError as err:
logger.error("Could not open requirements file: %s", err)
return 1
logger.debug("Wrote packages to %s", args.freeze_file)

joined_pip_cmd: str = " ".join(PIP_CMD)
joined_uninstall_args: str = " ".join(uninstall_args)
Expand All @@ -311,7 +321,7 @@ def main( # pylint: disable=R0914, R0915 # noqa: PLR0915
f"{running}: '{joined_pip_cmd} uninstall {joined_uninstall_args} "
f"{joined_packages_to_uninstall}'"
)
_logger.info(msg)
logger.info(msg)

if not args.dry_run:
uninstall_packages(
Expand Down
Loading
Loading