Skip to content

Commit

Permalink
Merge pull request #11 from realshouzy/improve-logging
Browse files Browse the repository at this point in the history 
Improve logging
  • Loading branch information
@realshouzy
realshouzy authored Jun 10, 2024
2 parents 7373979 + 6635adc commit c83325a
Show file tree
Hide file tree
Showing 8 changed files with 427 additions and 148 deletions.
82 changes: 61 additions & 21 deletions pip_manage/_logging.py
View file
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
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

0 comments on commit c83325a

Please sign in to comment.