|
3 | 3 | from collections.abc import Callable, Iterator |
4 | 4 | from contextlib import AbstractContextManager, contextmanager |
5 | 5 | from dataclasses import dataclass |
| 6 | +import logging |
| 7 | +from pathlib import Path |
| 8 | +import shutil |
| 9 | +import tempfile |
6 | 10 | import time |
| 11 | +from types import TracebackType |
| 12 | +import weakref as _weakref |
| 13 | + |
| 14 | +# Type aliases for exception handling |
| 15 | +type ExcType = type[BaseException] | None |
| 16 | +type ExcVal = BaseException | None |
| 17 | +type ExcTback = TracebackType | None |
| 18 | + |
| 19 | +# Logging |
| 20 | +log = logging.getLogger(__name__) |
7 | 21 |
|
8 | 22 |
|
9 | 23 | @dataclass |
@@ -53,3 +67,63 @@ def wrapper() -> Iterator[TimerData]: |
53 | 67 | data.elapsed = data.end - data.start |
54 | 68 |
|
55 | 69 | return wrapper |
| 70 | + |
| 71 | + |
| 72 | +class PersistentOnErrorTemporaryDirectory(tempfile.TemporaryDirectory): |
| 73 | + """Delete temporary directory only if no exception occurs. |
| 74 | +
|
| 75 | + It is similar to :class:`tempfile.TemporaryDirectory`, but it does not |
| 76 | + delete the directory if an exception occurs. This is useful for debugging |
| 77 | + or when you want to inspect the contents of the directory after an error. |
| 78 | +
|
| 79 | + .. code-block:: python |
| 80 | +
|
| 81 | + with PersistentOnErrorTemporaryDirectory() as temp_dir: |
| 82 | + # Do something with the temporary directory, it's a Path object |
| 83 | +
|
| 84 | + Optional arguments: |
| 85 | + :param suffix: A str suffix for the directory name. (see mkdtemp) |
| 86 | + :param prefix: A str prefix for the directory name. (see mkdtemp) |
| 87 | + :param dir: A directory to create this temp dir in. (see mkdtemp) |
| 88 | + """ |
| 89 | + |
| 90 | + def __init__( |
| 91 | + self, |
| 92 | + suffix: str | None = None, |
| 93 | + prefix: str | None = None, |
| 94 | + dir: str | Path | None = None, # noqa: A002 |
| 95 | + ) -> None: |
| 96 | + # Call the parent constructor. We don't need the |
| 97 | + # `ignore_cleanup_errors` flag as we implement our own cleanup. |
| 98 | + super().__init__(suffix=suffix, prefix=prefix, dir=dir) |
| 99 | + |
| 100 | + def __enter__(self) -> Path: |
| 101 | + """Enter the runtime context and create the temporary directory. |
| 102 | +
|
| 103 | + :returns: Path to the created temporary directory. |
| 104 | + """ |
| 105 | + # The parent __enter__ returns a string, so we override it |
| 106 | + # to return a Path object for consistency with your original class. |
| 107 | + return Path(self.name) |
| 108 | + |
| 109 | + def __exit__(self, exc_type: ExcType, exc_val: ExcVal, exc_tb: ExcTback) -> None: |
| 110 | + """Exit the runtime context and delete the directory if no exception occurred. |
| 111 | +
|
| 112 | + :param exc_type: Exception type, if any. |
| 113 | + :param exc_val: Exception instance, if any. |
| 114 | + :param exc_tb: Traceback, if any. |
| 115 | + """ |
| 116 | + # CRITICAL: We must always detach the finalizer. If we don't, |
| 117 | + # and an error occurred, the directory would still be deleted |
| 118 | + # upon garbage collection, which is not what we want. |
| 119 | + self._finalizer.detach() |
| 120 | + |
| 121 | + if exc_type is None: |
| 122 | + # No exception occurred in the `with` block, so we clean up. |
| 123 | + try: |
| 124 | + shutil.rmtree(self.name) |
| 125 | + |
| 126 | + except OSError as e: |
| 127 | + # Your custom logging is more informative than the parent's |
| 128 | + # `ignore_errors=True`, so we replicate it here. |
| 129 | + log.exception('Failed to delete temp dir %s: %s', self.name, e) |
0 commit comments