API Reference

Core Components

latenpy.latent(func=None, *, disable_cache=False)

Decorator to create a latent computation that executes only when requested.

Parameters:

• func (callable or None) – The function to be delayed. Will be None if decorator is called with parameters.

• disable_cache (bool, optional) – If True, disables caching of computation results. Each call to compute() will re-execute the function, by default False.

Returns:

If used as @delayed:

Returns a Latent object holding the function and arguments for later execution.

If used as @delayed(no_cache=…):

Returns a decorator function that will create a Latent object.

Return type:

callable or Latent

See also

Latent

The class that handles lazy evaluation of functions

Examples

Basic usage with default caching:

>>> @delayed
... def expensive_computation(x):
...     return x * 2
...
>>> result = expensive_computation(10)  # No computation yet
>>> result.compute()  # Now computes
20

Disable caching for always-fresh results:

>>> @delayed(no_cache=True)
... def always_recompute(x):
...     return x * 2
...
>>> result = always_recompute(10)
>>> result.compute()  # Computes without caching
20

Notes

The decorated function’s computation is deferred until the .compute() method is called on the returned Latent object. By default, results are cached based on input arguments unless no_cache=True.

class latenpy.Latent(func: Callable[[...], T], *args: Tuple[Any, ...], disable_cache: bool = False, **kwargs: Dict[str, Any])

Bases: object

A class for lazy evaluation with dependency tracking and caching.

This class enables delayed computation of functions and their arguments, with support for dependency tracking, result caching, and nested computations.

Parameters:

• func (Callable[..., T]) – The function to be computed lazily.

• *args (Tuple[Any, ...]) – Positional arguments for the function.

• disable_cache (bool, optional) – If True, disables result caching, by default False.

• **kwargs (Dict[str, Any]) – Keyword arguments for the function.

latent_data

Storage for cached computation results.

Type:

LatentData

computed

Whether the computation has been performed and cached.

Type:

bool

compute(force_recompute=False, recompute_dependencies=False, dont_cache=False)

Execute the computation and return the result.

update_func(func)

Update the function to be computed.

update_args(*args)

Update the positional arguments.

update_kwargs(**kwargs)

Update the keyword arguments.

clear_cache(dependencies=False, dependents=False)

Clear all cached results in the computation graph.

__init__(func: Callable[[...], T], *args: Tuple[Any, ...], disable_cache: bool = False, **kwargs: Dict[str, Any])

property args: Tuple[Any, ...]

Get the positional arguments for the function.

Returns:

The tuple of positional arguments.

Return type:

Tuple[Any, …]

clear_cache(dependencies: bool = False, dependents: bool = False) → None

Clear all cached results in this computation graph.

Parameters:

• dependencies (bool, optional) – If True, clear caches of dependencies, by default False

• dependents (bool, optional) – If True, clear caches of dependents, by default False

compute(force_recompute: bool = False, recompute_dependencies: bool = False, dont_cache: bool = False, maximum_depth: int = None) → T

Compute the result and cache if enabled.

Will iteratively compute all dependencies in arguments and key-word arguments if they are also Latent objects.

Parameters:

• force_recompute (bool, optional) – If True, recompute even if cached, by default False.

• recompute_dependencies (bool, optional) – If True, recompute all dependencies, by default False.

• dont_cache (bool, optional) – If True, skip caching the result, by default False.

• maximum_depth (int, optional) – Maximum depth for nested computations, by default None.

Returns:

The computed result.

Return type:

T

Raises:

• RecursionError – If a circular dependency is detected.

• Exception – If the computation fails, with details about the failure.

Notes

This method will: 1. Check for circular dependencies 2. Validate the dependency graph 3. Compute any required dependencies 4. Execute the computation 5. Cache the result (unless disabled)

property computed: bool

Check if the computation has been performed and cached.

Returns:

True if the result has been computed and cached, False otherwise.

Return type:

bool

property func: Callable[[...], T]

Get the function to be computed.

Returns:

The function that will be executed when compute() is called.

Return type:

Callable[…, T]

get_dependency_graph() → DiGraph

Build and return a directed graph of computation dependencies.

Returns:

A directed graph where: - Nodes are computations - Edges represent dependencies - Node attributes include:

• ’label’: Description of the computation

• ’computed’: Boolean for cache status

• ’func_name’: Name of the function

Return type:

DiGraph

property kwargs: Mapping[str, Any]

Get the keyword arguments for the function.

Returns:

The mapping of keyword arguments.

Return type:

Mapping[str, Any]

update_args(*args: Tuple[Any, ...]) → None

Update the positional arguments.

Parameters:

*args (Tuple[Any, ...]) – The new positional arguments.

Notes

This will clear the cached result and mark all dependent computations for recomputation.

update_func(func: Callable[[...], T]) → None

Update the function to be computed.

Parameters:

func (Callable[..., T]) – The new function to use for computation.

Notes

This will clear the cached result and mark all dependent computations for recomputation.

update_kwargs(full_reset: bool = False, **kwargs: Dict[str, Any]) → None

Update the keyword arguments.

Parameters:

• full_reset (bool, optional) – If True, replace all existing kwargs. If False, update only the provided kwargs, by default False.

• **kwargs (Dict[str, Any]) – The new keyword arguments.

Notes

This will clear the cached result and mark all dependent computations for recomputation.

Graph Utilities

latenpy.graph.analyze_dependencies(G: DiGraph) → Dict[str, Any]

Analyze the computation graph’s dependencies and structure.

Parameters:

G (DiGraph) – The directed graph to analyze.

Returns:

A dictionary containing the following metrics:

• depthint

  Maximum depth of the computation graph

• n_nodesint

  Total number of computation nodes

• n_edgesint

  Total number of dependencies

• leaf_nodesint

  Number of nodes with no dependencies

• root_nodesint

  Number of nodes with no dependents

• is_cyclicbool

  Whether the graph contains cycles

• max_in_degreeint

  Maximum number of direct dependencies for any node

• max_out_degreeint

  Maximum number of direct dependents for any node

Return type:

Dict[str, Any]

latenpy.graph.correct_computed_status(G: DiGraph) → None

Clear data from nodes that depend on any node that needs recomputing.

latenpy.graph.get_computed_nodes(G: DiGraph) → Set[str]

Return set of node IDs that have been computed.

latenpy.graph.get_optimized_pos(G: DiGraph, scale: float = 1.0)

Get optimized node positions with proper scaling.

latenpy.graph.get_uncached_nodes(G: DiGraph) → Set[str]

latenpy.graph.get_updated_nodes(G: DiGraph) → Set[str]

latenpy.graph.validate_no_cycles(G: DiGraph) → None

Validate that the computation graph has no cycles.

Raises:

ValueError – If cycles are detected in the dependency graph.

latenpy.graph.visualize(G: DiGraph, figsize: Tuple[int] = (8, 7), scale: float = 1.0, jitter: float = 0.0)

Types

class latenpy.types.LatentData

Bases: Generic[T]

Manages the state and metadata of a cached computation result.

data

The computed data.

Type:

T

computed

Whether the data has been computed.

Type:

bool

compute_time

The time of the last computation.

Type:

datetime

compute_count

The number of times the data has been computed.

Type:

int

last_access

The time of the last access.

Type:

datetime

access_count

The number of times the data has been accessed.

Type:

int

Examples

>>> latent_data = LatentData()
>>> latent_data.set(42)
>>> latent_data.get()
42
>>> if latent_data:
...     print("Data is computed")
Data is computed
>>> latent_data.stats
{
    "computed": True,
    "compute_count": 1,
    "access_count": 1,
    "last_compute": datetime.datetime(2021, 8, 24, 12, 0, 0),
    "last_access": datetime.datetime(2021, 8, 24, 12, 0, 0),
    "age": 0.0,
}
>>> latent_data.clear()
>>> if not latent_data:
...     print("Data is not computed")
Data is not computed
>>> latent_data.stats
{
    "computed": False,
    "compute_count": 0,
    "access_count": 0,
    "last_compute": None,
    "last_access": None,
    "age": None,
}

property age: float | None

Time since last computation in seconds.

clear(reset_stats: bool = False) → None

Clear the cached data, optionally resetting statistics.

get() → T

Get the computed data, updating access metadata.

set(data: T) → None

Set the computed data with metadata.

property stats: dict

Return computation statistics.

class latenpy.types.NotComputed

Bases: object

Singleton class to represent a computation that has not yet been performed.