Skip to content

impure

impure #

ImpureState — a Struct whose values persist per key and whose methods evolve them.

open(key) names one persistent state. Each method call on that handle locks the key, loads the committed state (or seeds one from the field defaults) as a writable copy, runs the method on it, and commits the copy as the new version.

PRIOR_STEM = 'prior' module-attribute #

The workspace subtree holding a method's writable copy of the committed state.

STATE_STEM = 'state' module-attribute #

The workspace subtree a new version is written to before sealing.

ImpureState #

Bases: Struct

Subclass with fields (the state) and public methods (its operations).

Method calls are never memoized: every call runs against the latest committed state. The identity is the class's qualified name plus its schema, so editing a method keeps the state and changing the fields starts fresh.

Source code in capturegraph-lib/capturegraph/recipes/calls/impure.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class ImpureState(Struct):
    """Subclass with fields (the state) and public methods (its operations).

    Method calls are never memoized: every call runs against the latest committed
    state. The identity is the class's qualified name plus its schema, so editing a
    method keeps the state and changing the fields starts fresh.
    """

    if TYPE_CHECKING:
        # A real class-body annotation would register as a Struct field.
        _cg_identity: ClassVar[str]

    def __init_subclass__(cls, **kwargs: object) -> None:
        """Register the fields as a Struct and derive the state identity."""
        if "__init__" in cls.__dict__:
            raise TypeError(
                f"{cls.__qualname__}: do not define __init__ on an ImpureState — "
                f"loading uses the dataclass constructor; seed with field defaults."
            )
        super().__init_subclass__(**kwargs)
        for name in cls._fields:
            if isinstance(cls.__dict__.get(name), FunctionType):
                raise TypeError(
                    f"{cls.__qualname__}.{name} is both a field and a method — rename one."
                )
        digest = hash_bytes(json.dumps(cls.schema(), sort_keys=True).encode()).hex
        cls._cg_identity = f"{cls.__module__}.{cls.__qualname__}@{digest}"

    @classmethod
    def open(cls, key: str = "default") -> Self:
        """A handle on the persistent state named ``key``; read it with ``peek``."""
        return cast(Self, _Handle(cls, key))

    def peek(self) -> Self:
        """The current committed state as a plain value (through a handle only)."""
        raise TypeError(
            f"{type(self).__name__}.peek() reads through a handle — "
            f"get one with {type(self).__name__}.open(key)."
        )

__init_subclass__(**kwargs) #

Register the fields as a Struct and derive the state identity.

Source code in capturegraph-lib/capturegraph/recipes/calls/impure.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init_subclass__(cls, **kwargs: object) -> None:
    """Register the fields as a Struct and derive the state identity."""
    if "__init__" in cls.__dict__:
        raise TypeError(
            f"{cls.__qualname__}: do not define __init__ on an ImpureState — "
            f"loading uses the dataclass constructor; seed with field defaults."
        )
    super().__init_subclass__(**kwargs)
    for name in cls._fields:
        if isinstance(cls.__dict__.get(name), FunctionType):
            raise TypeError(
                f"{cls.__qualname__}.{name} is both a field and a method — rename one."
            )
    digest = hash_bytes(json.dumps(cls.schema(), sort_keys=True).encode()).hex
    cls._cg_identity = f"{cls.__module__}.{cls.__qualname__}@{digest}"

open(key='default') classmethod #

A handle on the persistent state named key; read it with peek.

Source code in capturegraph-lib/capturegraph/recipes/calls/impure.py 
70
71
72
73
@classmethod
def open(cls, key: str = "default") -> Self:
    """A handle on the persistent state named ``key``; read it with ``peek``."""
    return cast(Self, _Handle(cls, key))

peek() #

The current committed state as a plain value (through a handle only).

Source code in capturegraph-lib/capturegraph/recipes/calls/impure.py 
75
76
77
78
79
80
def peek(self) -> Self:
    """The current committed state as a plain value (through a handle only)."""
    raise TypeError(
        f"{type(self).__name__}.peek() reads through a handle — "
        f"get one with {type(self).__name__}.open(key)."
    )

state_key(identity, shard) #

The scratch state key of one ImpureState class's shard.

Source code in capturegraph-lib/capturegraph/recipes/calls/impure.py 
37
38
39
def state_key(identity: str, shard: str) -> str:
    """The scratch state key of one ``ImpureState`` class's ``shard``."""
    return f"{identity}{KEY_SEPARATOR}{shard}"