feat: Allow configuring log file directory (#2687)

* feat: Allow configuring log file directory

Fixes #2398

Signed-off-by: Frost Ming <[email protected]>

* fix: catch_warnings

Signed-off-by: Frost Ming <[email protected]>

news/2398.feature.md

@@ -0,0 +1 @@
+Default to log to user home and make logs directory configurable.

pyproject.toml

@@ -212,8 +212,3 @@ addopts = "-r aR"
 testpaths = [
     "tests/",
 ]
-
-[tool.pyright]
-venvPath = "."
-venv = ".venv"
-pythonVersion = "3.8"

src/pdm/builders/base.py

@@ -19,7 +19,6 @@
 from pdm.models.requirements import Requirement, parse_requirement
 from pdm.models.working_set import WorkingSet
 from pdm.termui import logger
-from pdm.utils import create_tracked_tempdir
 
 if TYPE_CHECKING:
     from pdm.environments import BaseEnvironment
@@ -162,20 +161,18 @@ class EnvBuilder:
         _requires: list[str]
         _prefix: _Prefix
 
-    @classmethod
-    def get_shared_env(cls, key: int) -> str:
-        if key in cls._shared_envs:
-            logger.debug("Reusing shared build env: %s", cls._shared_envs[key])
-            return cls._shared_envs[key]
+    def get_shared_env(self, key: int) -> str:
+        if key in self._shared_envs:
+            logger.debug("Reusing shared build env: %s", self._shared_envs[key])
+            return self._shared_envs[key]
         # We don't save the cache here, instead it will be done after the installation
         # finished.
-        return create_tracked_tempdir("-shared", "pdm-build-env-")
+        return self._env.project.core.create_temp_dir("-shared", "pdm-build-env-")
 
-    @classmethod
-    def get_overlay_env(cls, key: str) -> str:
-        if key not in cls._overlay_envs:
-            cls._overlay_envs[key] = create_tracked_tempdir("-overlay", "pdm-build-env-")
-        return cls._overlay_envs[key]
+    def get_overlay_env(self, key: str) -> str:
+        if key not in self._overlay_envs:
+            self._overlay_envs[key] = self._env.project.core.create_temp_dir("-overlay", "pdm-build-env-")
+        return self._overlay_envs[key]
 
     def __init__(self, src_dir: str | Path, environment: BaseEnvironment) -> None:
         """If isolated is True(default), the builder will set up a *clean* environment.

src/pdm/core.py

@@ -5,15 +5,18 @@
  / ____/ /_/ / /  / /
 /_/   /_____/_/  /_/
 """
+
 from __future__ import annotations
 
 import argparse
+import contextlib
 import importlib
 import itertools
 import os
 import pkgutil
 import sys
 from pathlib import Path
+from tempfile import TemporaryDirectory
 from typing import TYPE_CHECKING, cast
 
 from resolvelib import Resolver
@@ -53,10 +56,14 @@ class Core:
 
     def __init__(self) -> None:
         self.version = __version__
-        self.ui = termui.UI()
+        self.exit_stack = contextlib.ExitStack()
+        self.ui = termui.UI(exit_stack=self.exit_stack)
         self.init_parser()
         self.load_plugins()
 
+    def create_temp_dir(self, *args: Any, **kwargs: Any) -> str:
+        return self.exit_stack.enter_context(TemporaryDirectory(*args, **kwargs))
+
     def init_parser(self) -> None:
         self.parser = ErrorArgumentParser(
             prog="pdm",
@@ -134,6 +141,8 @@ def handle(self, project: Project, options: argparse.Namespace) -> None:
 
         self.ui.set_verbosity(options.verbose)
         self.ui.set_theme(project.global_config.load_theme())
+        self.ui.log_dir = os.path.expanduser(cast(str, project.config["log_dir"]))
+        os.makedirs(self.ui.log_dir, exist_ok=True)
 
         command = cast("BaseCommand | None", getattr(options, "command", None))
         hooks = HookManager(project, getattr(options, "skip", None))
@@ -175,49 +184,49 @@ def main(
         **extra: Any,
     ) -> None:
         """The main entry function"""
-
-        if args is None:
-            args = []
-        args = self._get_cli_args(args, obj)
-        # Keep it for after project parsing to check if its a defined script
-        root_script = None
-        try:
-            options = self.parser.parse_args(args)
-        except PdmArgumentError as e:
-            # Failed to parse, try to give all to `run` command as shortcut
-            # and keep to root script (first non-dashed param) to check existence
-            # as soon as the project is parsed
-            root_script = next((arg for arg in args if not arg.startswith("-")), None)
-            if not root_script:
-                self.parser.error(str(e.__cause__))
+        with self.exit_stack:
+            if args is None:
+                args = []
+            args = self._get_cli_args(args, obj)
+            # Keep it for after project parsing to check if its a defined script
+            root_script = None
             try:
-                options = self.parser.parse_args(["run", *args])
+                options = self.parser.parse_args(args)
             except PdmArgumentError as e:
-                self.parser.error(str(e.__cause__))
-
-        project = self.ensure_project(options, obj)
-        if root_script and root_script not in project.scripts:
-            self.parser.error(f"Script unknown: {root_script}")
-
-        try:
-            self.handle(project, options)
-        except Exception:
-            etype, err, traceback = sys.exc_info()
-            should_show_tb = not isinstance(err, PdmUsageError)
-            if self.ui.verbosity > termui.Verbosity.NORMAL and should_show_tb:
-                raise cast(Exception, err).with_traceback(traceback) from None
-            self.ui.echo(
-                rf"[error]\[{etype.__name__}][/]: {err}",  # type: ignore[union-attr]
-                err=True,
-            )
-            if should_show_tb:
-                self.ui.warn("Add '-v' to see the detailed traceback", verbosity=termui.Verbosity.NORMAL)
-            sys.exit(1)
-        else:
-            if project.config["check_update"] and not is_in_zipapp():
-                from pdm.cli.actions import check_update
+                # Failed to parse, try to give all to `run` command as shortcut
+                # and keep to root script (first non-dashed param) to check existence
+                # as soon as the project is parsed
+                root_script = next((arg for arg in args if not arg.startswith("-")), None)
+                if not root_script:
+                    self.parser.error(str(e.__cause__))
+                try:
+                    options = self.parser.parse_args(["run", *args])
+                except PdmArgumentError as e:
+                    self.parser.error(str(e.__cause__))
+
+            project = self.ensure_project(options, obj)
+            if root_script and root_script not in project.scripts:
+                self.parser.error(f"Script unknown: {root_script}")
 
-                check_update(project)
+            try:
+                self.handle(project, options)
+            except Exception:
+                etype, err, traceback = sys.exc_info()
+                should_show_tb = not isinstance(err, PdmUsageError)
+                if self.ui.verbosity > termui.Verbosity.NORMAL and should_show_tb:
+                    raise cast(Exception, err).with_traceback(traceback) from None
+                self.ui.echo(
+                    rf"[error]\[{etype.__name__}][/]: {err}",  # type: ignore[union-attr]
+                    err=True,
+                )
+                if should_show_tb:
+                    self.ui.warn("Add '-v' to see the detailed traceback", verbosity=termui.Verbosity.NORMAL)
+                sys.exit(1)
+            else:
+                if project.config["check_update"] and not is_in_zipapp():
+                    from pdm.cli.actions import check_update
+
+                    check_update(project)
 
     def register_command(self, command: type[BaseCommand], name: str | None = None) -> None:
         """Register a subcommand to the subparsers,

src/pdm/installers/synchronizers.py

@@ -268,9 +268,7 @@ def synchronize(self) -> None:
 
 
 class Synchronizer(BaseSynchronizer):
-    def create_executor(
-        self,
-    ) -> ThreadPoolExecutor | DummyExecutor:
+    def create_executor(self) -> ThreadPoolExecutor | DummyExecutor:
         if self.parallel:
             return ThreadPoolExecutor(max_workers=min(multiprocessing.cpu_count(), 8))
         else:

src/pdm/models/caches.py

@@ -111,7 +111,6 @@ def _get_key(cls, obj: Candidate) -> str:
 
 
 class HashCache:
-
     """Caches hashes of PyPI artifacts so we do not need to re-download them.
 
     Hashes are only cached when the URL appears to contain a hash in it and the
@@ -324,7 +323,7 @@ def delete(self, key: str) -> None:
             os.remove(path)
 
 
-@lru_cache(maxsize=128)
+@lru_cache(maxsize=None)
 def get_wheel_cache(directory: Path | str) -> WheelCache:
     return WheelCache(directory)
 

src/pdm/models/candidates.py

@@ -33,7 +33,6 @@
     cd,
     comparable_version,
     convert_hashes,
-    create_tracked_tempdir,
     filtered_sources,
     get_file_hash,
     get_rev_from_url,
@@ -519,7 +518,7 @@ def prepare_metadata(self, force_build: bool = False) -> im.Distribution:
                 return dist
 
         # If all fail, try building the source to get the metadata
-        metadata_parent = create_tracked_tempdir(prefix="pdm-meta-")
+        metadata_parent = self.environment.project.core.create_temp_dir(prefix="pdm-meta-")
         return self._get_metadata_from_build(self._unpacked_dir, metadata_parent)
 
     def _get_metadata_from_metadata_link(
@@ -691,7 +690,7 @@ def _get_build_dir(self) -> str:
                 dirname, _ = os.path.splitext(original_link.filename)
             return str(src_dir / str(dirname))
         # Otherwise, for source dists, they will be unpacked into a *temp* directory.
-        return create_tracked_tempdir(prefix="pdm-build-")
+        return self.environment.project.core.create_temp_dir(prefix="pdm-build-")
 
     def _wheel_compatible(self, wheel_file: str, allow_all: bool = False) -> bool:
         if allow_all:

src/pdm/project/config.py

@@ -101,6 +101,12 @@ class Config(MutableMapping[str, str]):
             True,
             env_var="PDM_CACHE_DIR",
         ),
+        "log_dir": ConfigItem(
+            "The root directory of log files",
+            platformdirs.user_log_dir("pdm"),
+            True,
+            env_var="PDM_LOG_DIR",
+        ),
         "check_update": ConfigItem(
             "Check if there is any newer version available",
             True,

src/pdm/pytest.py

@@ -26,7 +26,6 @@
 import os
 import shutil
 import sys
-import warnings
 from dataclasses import dataclass
 from io import BufferedReader, BytesIO, StringIO
 from pathlib import Path
@@ -372,7 +371,7 @@ def core() -> Iterator[Core]:
     # Turn off use_venv by default, for testing
     Config._config_map["python.use_venv"].default = False
     main = Core()
-    with warnings.catch_warnings():
+    with main.exit_stack:
         yield main
     # Restore the config items
     Config._config_map = old_config_map
@@ -404,6 +403,7 @@ def project_no_init(
     mocker.patch("pdm.builders.base.EnvBuilder.get_shared_env", return_value=str(build_env))
     tmp_path.joinpath("caches").mkdir(parents=True)
     p.global_config["cache_dir"] = tmp_path.joinpath("caches").as_posix()
+    p.global_config["log_dir"] = tmp_path.joinpath("logs").as_posix()
     python_path = find_python_in_path(sys.base_prefix)
     if python_path is None:
         raise ValueError("Unable to find a Python path")

src/pdm/termui.py

@@ -1,12 +1,11 @@
 from __future__ import annotations
 
-import atexit
 import contextlib
 import enum
 import logging
 import os
+import tempfile
 import warnings
-from tempfile import mktemp
 from typing import TYPE_CHECKING
 
 from rich.box import ROUNDED
@@ -158,12 +157,17 @@ def _show(self) -> None:
 class UI:
     """Terminal UI object"""
 
-    def __init__(self, verbosity: Verbosity = Verbosity.NORMAL) -> None:
+    def __init__(
+        self, verbosity: Verbosity = Verbosity.NORMAL, *, exit_stack: contextlib.ExitStack | None = None
+    ) -> None:
         self.verbosity = verbosity
+        self.exit_stack = exit_stack or contextlib.ExitStack()
+        self.log_dir: str | None = None
 
     def set_verbosity(self, verbosity: int) -> None:
         self.verbosity = Verbosity(verbosity)
         if self.verbosity == Verbosity.QUIET:
+            self.exit_stack.enter_context(warnings.catch_warnings())
             warnings.simplefilter("ignore", PDMWarning, append=True)
             warnings.simplefilter("ignore", FutureWarning, append=True)
 
@@ -226,37 +230,38 @@ def logging(self, type_: str = "install") -> Iterator[logging.Logger]:
         """A context manager that opens a file for logging when verbosity is NORMAL or
         print to the stdout otherwise.
         """
-        file_name: str | None = None
+        log_file: str | None = None
         if self.verbosity >= Verbosity.DETAIL:
             handler: logging.Handler = logging.StreamHandler()
             handler.setLevel(LOG_LEVELS[self.verbosity])
         else:
-            file_name = mktemp(".log", f"pdm-{type_}-")
-            handler = logging.FileHandler(file_name, encoding="utf-8")
+            log_file = tempfile.mktemp(".log", f"pdm-{type_}-", self.log_dir)
+            handler = logging.FileHandler(log_file, encoding="utf-8")
             handler.setLevel(logging.DEBUG)
+
         handler.setFormatter(logging.Formatter("%(name)s: %(message)s"))
         logger.addHandler(handler)
         unearth_logger.addHandler(handler)
 
         def cleanup() -> None:
-            if not file_name:
+            if not log_file:
                 return
             with contextlib.suppress(OSError):
-                os.unlink(file_name)
+                os.unlink(log_file)
 
         try:
             yield logger
         except Exception:
             if self.verbosity < Verbosity.DETAIL:
                 logger.exception("Error occurs")
                 self.echo(
-                    f"See [warning]{file_name}[/] for detailed debug log.",
+                    f"See [warning]{log_file}[/] for detailed debug log.",
                     style="error",
                     err=True,
                 )
             raise
         else:
-            atexit.register(cleanup)
+            self.exit_stack.callback(cleanup)
         finally:
             logger.removeHandler(handler)
             unearth_logger.removeHandler(handler)

tests/models/test_candidates.py

@@ -89,14 +89,6 @@ def test_parse_remote_link_metadata(project):
     assert candidate.version == "0.0.1"
 
 
-@pytest.mark.xfail(reason="packaging 22 no longer supports legacy specifiers")
-@pytest.mark.usefixtures("local_finder")
-def test_parse_abnormal_specifiers(project):
-    req = parse_requirement("http://fixtures.test/artifacts/celery-4.4.2-py2.py3-none-any.whl")
-    candidate = Candidate(req)
-    assert candidate.prepare(project.environment).get_dependencies_from_metadata()
-
-
 @pytest.mark.usefixtures("local_finder")
 @pytest.mark.parametrize(
     "req_str",

tests/test_project.py

@@ -389,3 +389,16 @@ def expect_sources(requirement: str, expected: list[str]) -> bool:
     expect_sources("foo-bar", ["source1", "source2"])
     expect_sources("bar-extra", ["source2"])
     expect_sources("baz-extra", ["source1", "pypi"])
+
+
+@pytest.mark.usefixtures("working_set")
+def test_preserve_log_file(project, pdm, tmp_path, mocker):
+    pdm(["add", "requests"], obj=project, strict=True)
+    all_logs = list(tmp_path.joinpath("logs").iterdir())
+    assert len(all_logs) == 0
+
+    with mocker.patch.object(project.core.synchronizer_class, "synchronize", side_effect=Exception):
+        result = pdm(["add", "pytz"], obj=project)
+        assert result.exit_code != 0
+        install_log = next(tmp_path.joinpath("logs").glob("pdm-install-*.log"))
+        assert install_log.exists()