Skip to content

plaid.viewer.cache

plaid.viewer.cache

Ephemeral artifact cache for the dataset viewer.

The cache lives under a per-process temporary directory and is removed at shutdown. Four cleanup layers cover all practical failure modes:

  1. atexit.register for normal Python exit.
  2. Signal handlers for SIGINT / SIGTERM.
  3. A context manager (with CacheRoot() as cache: in the CLI).
  4. An orphan sweep at startup that removes directories left behind by previously-crashed processes.

plaid.viewer.cache.CacheRoot 

CacheRoot(
    *, install_signal_handlers=True, run_orphan_sweep=True
)

Context-manager-friendly ephemeral artifact cache directory.

Creates a new tempdir named plaid-viewer-{pid}-{token} under the OS temp root. The directory is removed at process exit (atexit), on SIGINT / SIGTERM, and when the context manager is closed.

Source code in plaid/viewer/cache.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def __init__(
    self,
    *,
    install_signal_handlers: bool = True,
    run_orphan_sweep: bool = True,
) -> None:
    if run_orphan_sweep:
        sweep_orphans()
    token = uuid.uuid4().hex[:12]
    base = Path(tempfile.gettempdir())
    self._path = base / f"{_EPHEMERAL_PREFIX}{os.getpid()}-{token}"
    self._path.mkdir(parents=True, exist_ok=False)
    atexit.register(self._safe_cleanup)
    if install_signal_handlers:
        self._install_signal_handlers()
    self._closed = False

plaid.viewer.cache.CacheRoot.path property 

path

Root directory of the cache.

plaid.viewer.cache.CacheRoot.close 

close()

Remove the cache directory.

Source code in plaid/viewer/cache.py 
139
140
141
142
143
144
def close(self) -> None:
    """Remove the cache directory."""
    if self._closed:
        return
    self._closed = True
    self._safe_cleanup()

plaid.viewer.cache.sweep_orphans 

sweep_orphans(temp_root=None)

Remove viewer tempdirs whose owning process is no longer running.

Parameters:

  • temp_root (Path | None, default: None ) –

    Base temp directory to scan. Defaults to :func:tempfile.gettempdir.

Returns:

  • list[Path] –

    List of directories that were removed.

Source code in plaid/viewer/cache.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def sweep_orphans(temp_root: Path | None = None) -> list[Path]:
    """Remove viewer tempdirs whose owning process is no longer running.

    Args:
        temp_root: Base temp directory to scan. Defaults to
            :func:`tempfile.gettempdir`.

    Returns:
        List of directories that were removed.
    """
    root = Path(temp_root) if temp_root is not None else Path(tempfile.gettempdir())
    removed: list[Path] = []
    if not root.is_dir():
        return removed
    for entry in root.iterdir():
        if not entry.is_dir():
            continue
        match = _EPHEMERAL_PATTERN.match(entry.name)
        if match is None:
            continue
        pid = int(match.group("pid"))
        if _process_is_alive(pid):
            continue
        try:
            shutil.rmtree(entry, ignore_errors=True)
            removed.append(entry)
            logger.info("Removed orphan viewer cache: %s", entry)
        except OSError as exc:
            logger.warning("Could not remove orphan viewer cache %s: %s", entry, exc)
    return removed