teff.checkpoint.base

teff.checkpoint.base

Checkpointing primitives for durable graph execution.

Classes:

Name	Description
Checkpoint	A snapshot of execution that can be resumed from.
Checkpointer	Interface for persisting execution checkpoints.

Functions:

Name	Description
checkpoint_from_dict	Reconstruct a checkpoint from a dict produced by :func:checkpoint_to_dict.
checkpoint_to_dict	Convert a checkpoint to a JSON-serializable dict.

Checkpoint dataclass

A snapshot of execution that can be resumed from.

Attributes:

Name	Type	Description
state	dict	Workflow state data (JSON-serializable dict).
next_node_id	str | None	ID of the node to execute on resume. None means the graph completed.
iteration	int	Number of completed node executions so far.

Source code in teff/checkpoint/base.py

@dataclass
class Checkpoint:
    """A snapshot of execution that can be resumed from.

    Attributes:
        state: Workflow state data (JSON-serializable dict).
        next_node_id: ID of the node to execute on resume.
            ``None`` means the graph completed.
        iteration: Number of completed node executions so far.
    """

    state: dict
    next_node_id: str | None
    iteration: int

Checkpointer

Bases: Protocol

Interface for persisting execution checkpoints.

Implementations must be safe to call concurrently for different checkpoint IDs and must persist atomically enough that a crash never leaves a partially-written checkpoint.

The optional owner scopes a checkpoint to a user/session/tenant. Two checkpoints with the same ID but different owners never collide. When omitted the default owner (:data:DEFAULT_OWNER, "default") is used, keeping single-tenant callers unchanged.

Use a distinct owner per end-user (e.g. a user id or session id) so every tenant's runs are isolated from the others and can be listed with :meth:list.

Methods:

Name	Description
cleanup	Delete stale checkpoints; returns how many were removed.
delete	Remove a saved checkpoint. No-op if it does not exist.
list	Return all checkpoint IDs persisted for owner.
load	Return the saved checkpoint for owner, or None if never saved.
save	Persist checkpoint under checkpoint_id for owner.

Source code in teff/checkpoint/base.py

class Checkpointer(Protocol):
    """Interface for persisting execution checkpoints.

    Implementations must be safe to call concurrently for different
    checkpoint IDs and must persist *atomically* enough that a crash
    never leaves a partially-written checkpoint.

    The optional *owner* scopes a checkpoint to a user/session/tenant.
    Two checkpoints with the same ID but different owners never collide.
    When omitted the default owner (:data:`DEFAULT_OWNER`, ``"default"``)
    is used, keeping single-tenant callers unchanged.

    Use a distinct owner per end-user (e.g. a user id or session id) so
    every tenant's runs are isolated from the others and can be listed
    with :meth:`list`.
    """

    async def save(
        self, checkpoint_id: str, checkpoint: Checkpoint, *, owner: str = DEFAULT_OWNER
    ) -> None:
        """Persist *checkpoint* under *checkpoint_id* for *owner*."""
        ...

    async def load(
        self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER
    ) -> Checkpoint | None:
        """Return the saved checkpoint for *owner*, or ``None`` if never saved."""
        ...

    async def delete(self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER) -> None:
        """Remove a saved checkpoint. No-op if it does not exist."""
        ...

    async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
        """Return all checkpoint IDs persisted for *owner*."""
        ...

    async def cleanup(
        self,
        *,
        owner: str | None = None,
        max_age: float | None = None,
        keep_last: int | None = None,
    ) -> int:
        """Delete stale checkpoints; returns how many were removed.

        ``owner=None`` cleans up every owner; otherwise only that owner.
        ``max_age`` removes checkpoints last written more than that many
        seconds ago.  ``keep_last`` retains the *N* most recently written
        checkpoints per owner (after any ``max_age`` pruning) and deletes
        the rest.  When both are omitted nothing is deleted.
        """
        ...

cleanup async

cleanup(*, owner=None, max_age=None, keep_last=None)

Delete stale checkpoints; returns how many were removed.

owner=None cleans up every owner; otherwise only that owner. max_age removes checkpoints last written more than that many seconds ago. keep_last retains the N most recently written checkpoints per owner (after any max_age pruning) and deletes the rest. When both are omitted nothing is deleted.

Source code in teff/checkpoint/base.py

async def cleanup(
    self,
    *,
    owner: str | None = None,
    max_age: float | None = None,
    keep_last: int | None = None,
) -> int:
    """Delete stale checkpoints; returns how many were removed.

    ``owner=None`` cleans up every owner; otherwise only that owner.
    ``max_age`` removes checkpoints last written more than that many
    seconds ago.  ``keep_last`` retains the *N* most recently written
    checkpoints per owner (after any ``max_age`` pruning) and deletes
    the rest.  When both are omitted nothing is deleted.
    """
    ...

delete async

delete(checkpoint_id, *, owner=DEFAULT_OWNER)

Remove a saved checkpoint. No-op if it does not exist.

Source code in teff/checkpoint/base.py

async def delete(self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER) -> None:
    """Remove a saved checkpoint. No-op if it does not exist."""
    ...

list async

list(owner=DEFAULT_OWNER)

Return all checkpoint IDs persisted for owner.

Source code in teff/checkpoint/base.py

async def list(self, owner: str = DEFAULT_OWNER) -> list[str]:
    """Return all checkpoint IDs persisted for *owner*."""
    ...

load async

load(checkpoint_id, *, owner=DEFAULT_OWNER)

Return the saved checkpoint for owner, or None if never saved.

Source code in teff/checkpoint/base.py

async def load(
    self, checkpoint_id: str, *, owner: str = DEFAULT_OWNER
) -> Checkpoint | None:
    """Return the saved checkpoint for *owner*, or ``None`` if never saved."""
    ...

save async

save(checkpoint_id, checkpoint, *, owner=DEFAULT_OWNER)

Persist checkpoint under checkpoint_id for owner.

Source code in teff/checkpoint/base.py

async def save(
    self, checkpoint_id: str, checkpoint: Checkpoint, *, owner: str = DEFAULT_OWNER
) -> None:
    """Persist *checkpoint* under *checkpoint_id* for *owner*."""
    ...

checkpoint_from_dict

checkpoint_from_dict(data)

Reconstruct a checkpoint from a dict produced by :func:checkpoint_to_dict.

Source code in teff/checkpoint/base.py

def checkpoint_from_dict(data: dict[str, Any]) -> Checkpoint:
    """Reconstruct a checkpoint from a dict produced by :func:`checkpoint_to_dict`."""
    return Checkpoint(
        state=data.get("state", {}),
        next_node_id=data.get("next_node_id"),
        iteration=data.get("iteration", 0),
    )

checkpoint_to_dict

checkpoint_to_dict(cp)

Convert a checkpoint to a JSON-serializable dict.

Source code in teff/checkpoint/base.py

def checkpoint_to_dict(cp: Checkpoint) -> dict[str, Any]:
    """Convert a checkpoint to a JSON-serializable dict."""
    return {
        "state": cp.state,
        "next_node_id": cp.next_node_id,
        "iteration": cp.iteration,
    }