Skip to content

workspace

workspace #

A live workspace: a directory under live/<owner>/ that lasts as long as it is held.

LOG_NAME = 'exec.log' module-attribute #

A call's log, at the workspace root and so outside every result that seals.

LOG_TAIL_LINES = 40 module-attribute #

How much of a log a failure note or a progress display carries.

Workspace #

A fresh directory a call writes into, deleted when the last holder lets go.

Source code in capturegraph-lib/capturegraph/recipes/scratch/store/workspace.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class Workspace:
    """A fresh directory a call writes into, deleted when the last holder lets go."""

    __slots__ = ("__weakref__", "_finalizer", "path")

    def __init__(self, live_dir: Path) -> None:
        """Create the next workspace directory under this owner's ``live_dir``."""
        self.path = live_dir / f"{next(_counter):x}"
        self.path.mkdir(parents=True)
        self._finalizer = weakref.finalize(self, remove_tree, self.path)

    @property
    def log(self) -> Path:
        """Where a tool run in this workspace writes its combined output."""
        return self.path / LOG_NAME

    def release(self) -> None:
        """Delete the directory now."""
        self._finalizer()

    def __enter__(self) -> Workspace:
        """Hold the workspace for the block."""
        return self

    def __exit__(self, *_: object) -> None:
        """Release the workspace."""
        self.release()

    def __repr__(self) -> str:
        """``Workspace(<path>)``."""
        return f"Workspace({str(self.path)!r})"

log property #

Where a tool run in this workspace writes its combined output.

__enter__() #

Hold the workspace for the block.

Source code in capturegraph-lib/capturegraph/recipes/scratch/store/workspace.py 
42
43
44
def __enter__(self) -> Workspace:
    """Hold the workspace for the block."""
    return self

__exit__(*_) #

Release the workspace.

Source code in capturegraph-lib/capturegraph/recipes/scratch/store/workspace.py 
46
47
48
def __exit__(self, *_: object) -> None:
    """Release the workspace."""
    self.release()

__init__(live_dir) #

Create the next workspace directory under this owner's live_dir.

Source code in capturegraph-lib/capturegraph/recipes/scratch/store/workspace.py 
27
28
29
30
31
def __init__(self, live_dir: Path) -> None:
    """Create the next workspace directory under this owner's ``live_dir``."""
    self.path = live_dir / f"{next(_counter):x}"
    self.path.mkdir(parents=True)
    self._finalizer = weakref.finalize(self, remove_tree, self.path)

__repr__() #

Workspace(<path>).

Source code in capturegraph-lib/capturegraph/recipes/scratch/store/workspace.py 
50
51
52
def __repr__(self) -> str:
    """``Workspace(<path>)``."""
    return f"Workspace({str(self.path)!r})"

release() #

Delete the directory now.

Source code in capturegraph-lib/capturegraph/recipes/scratch/store/workspace.py 
38
39
40
def release(self) -> None:
    """Delete the directory now."""
    self._finalizer()

log_tail(workspace, lines=LOG_TAIL_LINES) #

The last lines lines logged in the workspace directory workspace.

Parameters:

Name Type Description Default
workspace Path

A workspace directory, which may be gone already.

 required
lines int

How many trailing lines to read.

 LOG_TAIL_LINES

Returns:

Type Description
str | None

The tail, or None if no tool logged there.
Source code in capturegraph-lib/capturegraph/recipes/scratch/store/workspace.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def log_tail(workspace: Path, lines: int = LOG_TAIL_LINES) -> str | None:
    """The last ``lines`` lines logged in the workspace directory ``workspace``.

    Args:
        workspace: A workspace directory, which may be gone already.
        lines: How many trailing lines to read.

    Returns:
        The tail, or ``None`` if no tool logged there.
    """
    try:
        logged = (workspace / LOG_NAME).read_text(errors="replace")
    except OSError:
        return None
    return "\n".join(logged.splitlines()[-lines:])