Skip to content

workspace

workspace #

The ambient workspace a body runs in, and the holds keeping a value's files alive.

active_workspace = ContextVar('cg_active_workspace', default=None) module-attribute #

The live workspace of the recipe call running in this context, if any.

allocate(cls, suffix) #

A cls value at a fresh, not-yet-existing path in the ambient workspace.

Outside any call, a workspace is opened for this value alone and held by it.

Raises:

Type Description
ScratchNotConfigured

If no scratch is configured.

Source code in capturegraph-lib/capturegraph/recipes/values/workspace.py
def allocate[T: Path](cls: type[T], suffix: str) -> T:
    """A ``cls`` value at a fresh, not-yet-existing path in the ambient workspace.

    Outside any call, a workspace is opened for this value alone and held by it.

    Raises:
        ScratchNotConfigured: If no scratch is configured.
    """
    workspace = active_workspace.get()
    hold = None
    if workspace is None:
        workspace = hold = active_scratch().workspace()
    value = cls(workspace.path / f"{uuid.uuid4().hex}{suffix}")
    if hold is not None:
        pin(value, hold)
    return value

computing(scratch, key, identity, label) #

A fresh workspace to compute the call key in, ambient and reported as a step.

The block seals the workspace or loses it: on any exception the workspace is released, and the exception carries the tail of its exec.log as a note.

Yields:

Type Description
Workspace

The workspace the block computes in.

Source code in capturegraph-lib/capturegraph/recipes/values/workspace.py
@contextmanager
def computing(
    scratch: Scratch,
    key: CallKey,
    identity: str,
    label: str,
) -> Iterator[Workspace]:
    """A fresh workspace to compute the call ``key`` in, ambient and reported as a step.

    The block seals the workspace or loses it: on any exception the workspace is
    released, and the exception carries the tail of its ``exec.log`` as a note.

    Yields:
        The workspace the block computes in.
    """
    workspace = scratch.workspace()
    with opened(Step(label, identity=identity, call_key=key, workspace=workspace.path)):
        workspace_token = active_workspace.set(workspace)
        try:
            yield workspace
        except BaseException as error:
            tail = log_tail(workspace.path)
            if tail is not None:
                error.add_note(tail)
            workspace.release()
            raise
        finally:
            active_workspace.reset(workspace_token)

exec_log(value) #

Where a tool run by value logs, or None outside every workspace.

The ambient workspace's log, else that of the workspace value holds.

Source code in capturegraph-lib/capturegraph/recipes/values/workspace.py
def exec_log(value: object) -> Path | None:
    """Where a tool run by ``value`` logs, or ``None`` outside every workspace.

    The ambient workspace's log, else that of the workspace ``value`` holds.
    """
    workspace = active_workspace.get()
    if workspace is None:
        hold = pinned(value)
        workspace = hold if isinstance(hold, Workspace) else None
    return workspace.log if workspace is not None else None

pin(value, hold) #

Keep hold (a workspace or object pin) alive as long as value is.

Source code in capturegraph-lib/capturegraph/recipes/values/workspace.py
def pin(value: object, hold: object) -> None:
    """Keep ``hold`` (a workspace or object pin) alive as long as ``value`` is."""
    object.__setattr__(value, "_cg_pin", hold)

pinned(value) #

The hold value carries, if any.

Source code in capturegraph-lib/capturegraph/recipes/values/workspace.py
def pinned(value: object) -> object | None:
    """The hold ``value`` carries, if any."""
    return getattr(value, "_cg_pin", None)