Skip to content

pure

pure #

pure_function — memoize a pure function over CGType values.

PureFunction #

A memoized pure function: a plain synchronous call.

Source code in capturegraph-lib/capturegraph/recipes/calls/pure.py
class PureFunction[**P, T]:
    """A memoized pure function: a plain synchronous call."""

    def __init__(self, func: Callable[P, T], identity: str | None) -> None:
        """Wrap ``func``; see ``pure_function`` for the parameters.

        Args:
            func: The pure body to memoize.
            identity: An explicit versioned identity, or ``None`` for automatic.
        """
        self._func = func
        self._signature = inspect.signature(func)
        self._returns: ReturnSpec = return_spec(func)
        self._parameters = parameter_specs(func)
        self.identity = function_identity(func, identity)
        self.label = label_of(func)
        functools.update_wrapper(self, func)

    def key(self, *args: P.args, **kwargs: P.kwargs) -> CallKey:
        """The cache key of this function applied to these arguments, defaults applied.

        Raises:
            UnhashableArgumentError: If an argument has no canonical content hash.
        """
        bound = self._signature.bind(*args, **kwargs)
        bound.apply_defaults()
        return call_key(
            self.identity,
            [
                argument_hash(name, value, self._parameters.get(name))
                for name, value in bound.arguments.items()
            ],
        )

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> T:
        """Serve the call from cache, or run the body and seal the result.

        A ``Missing`` argument makes the result ``Missing`` without running or
        caching anything, so absence chains through a recipe as it does
        through attribute access.
        """
        if any(isinstance(value, _Missing) for value in (*args, *kwargs.values())):
            return cast(T, Missing)
        return cast(
            T,
            run_call(
                self.key(*args, **kwargs),
                self.identity,
                self.label,
                self._returns,
                lambda: self._func(*args, **kwargs),
            ),
        )

__call__(*args, **kwargs) #

Serve the call from cache, or run the body and seal the result.

A Missing argument makes the result Missing without running or caching anything, so absence chains through a recipe as it does through attribute access.

Source code in capturegraph-lib/capturegraph/recipes/calls/pure.py
def __call__(self, *args: P.args, **kwargs: P.kwargs) -> T:
    """Serve the call from cache, or run the body and seal the result.

    A ``Missing`` argument makes the result ``Missing`` without running or
    caching anything, so absence chains through a recipe as it does
    through attribute access.
    """
    if any(isinstance(value, _Missing) for value in (*args, *kwargs.values())):
        return cast(T, Missing)
    return cast(
        T,
        run_call(
            self.key(*args, **kwargs),
            self.identity,
            self.label,
            self._returns,
            lambda: self._func(*args, **kwargs),
        ),
    )

__init__(func, identity) #

Wrap func; see pure_function for the parameters.

Parameters:

Name Type Description Default
func Callable[P, T]

The pure body to memoize.

required
identity str | None

An explicit versioned identity, or None for automatic.

required
Source code in capturegraph-lib/capturegraph/recipes/calls/pure.py
def __init__(self, func: Callable[P, T], identity: str | None) -> None:
    """Wrap ``func``; see ``pure_function`` for the parameters.

    Args:
        func: The pure body to memoize.
        identity: An explicit versioned identity, or ``None`` for automatic.
    """
    self._func = func
    self._signature = inspect.signature(func)
    self._returns: ReturnSpec = return_spec(func)
    self._parameters = parameter_specs(func)
    self.identity = function_identity(func, identity)
    self.label = label_of(func)
    functools.update_wrapper(self, func)

key(*args, **kwargs) #

The cache key of this function applied to these arguments, defaults applied.

Raises:

Type Description
UnhashableArgumentError

If an argument has no canonical content hash.

Source code in capturegraph-lib/capturegraph/recipes/calls/pure.py
def key(self, *args: P.args, **kwargs: P.kwargs) -> CallKey:
    """The cache key of this function applied to these arguments, defaults applied.

    Raises:
        UnhashableArgumentError: If an argument has no canonical content hash.
    """
    bound = self._signature.bind(*args, **kwargs)
    bound.apply_defaults()
    return call_key(
        self.identity,
        [
            argument_hash(name, value, self._parameters.get(name))
            for name, value in bound.arguments.items()
        ],
    )

pure_function(identity=None) #

pure_function(
    identity: Callable[P, T],
) -> PureFunction[P, T]
pure_function(
    identity: str | None = None,
) -> Callable[[Callable[P, T]], PureFunction[P, T]]

Memoize a pure function over CGType values in the scratch cache.

Invalidation contract: a versioned identity ("name/1") means you own invalidation — editing the body without bumping the version keeps serving the stale cache, deliberately; bump to "name/2" to invalidate. The bare form digests the compiled body instead, so any edit invalidates automatically.

Parameters:

Name Type Description Default
identity Callable[P, T] | str | None

A versioned identity string, None for the automatic source-digest identity — or, in the bare form, the function itself.

None

Returns:

Type Description
PureFunction[P, T] | Callable[[Callable[P, T]], PureFunction[P, T]]

The memoized function, or a decorator producing one.

Raises:

Type Description
ValueError

If an explicit identity is not of the "name/N" shape.

TypeError

If the return annotation is missing or not CGType-typed.

Source code in capturegraph-lib/capturegraph/recipes/calls/pure.py
def pure_function[**P, T](
    identity: Callable[P, T] | str | None = None,
) -> PureFunction[P, T] | Callable[[Callable[P, T]], PureFunction[P, T]]:
    """Memoize a pure function over ``CGType`` values in the scratch cache.

    **Invalidation contract**: a versioned identity (``"name/1"``) means *you*
    own invalidation — editing the body without bumping the version keeps
    serving the stale cache, deliberately; bump to ``"name/2"`` to invalidate.
    The bare form digests the compiled body instead, so any edit invalidates
    automatically.

    Args:
        identity: A versioned identity string, ``None`` for the automatic
            source-digest identity — or, in the bare form, the function itself.

    Returns:
        The memoized function, or a decorator producing one.

    Raises:
        ValueError: If an explicit identity is not of the ``"name/N"`` shape.
        TypeError: If the return annotation is missing or not CGType-typed.
    """
    if callable(identity):
        return PureFunction(identity, None)
    decorated_identity = identity

    def decorate(func: Callable[P, T]) -> PureFunction[P, T]:
        return PureFunction(func, decorated_identity)

    return decorate