temporalio.contrib.langgraph._task_cache

temporalio.contrib.langgraph._task_cache

module documentation
(source)

Task result cache for continue-as-new support.

Caches task results by (module.qualname, args, kwargs) hash so that previously completed tasks are not re-executed after a continue-as-new. The cache state is a plain dict that can travel through workflow.continue_as_new().

def cache_key(task_id: str, args: tuple[Any, ...], kwargs: dict[str, Any], context: Any = None) -> str: (source)

Build a cache key from the full task identifier, arguments, and runtime context.

def cache_lookup(key: str) -> tuple[bool, Any]: (source)

Return (True, value) if cached, (False, None) otherwise.

def cache_put(key: str, value: Any): (source)

Store a value in the task result cache.

def get_task_cache() -> dict[str, Any] | None: (source)

Get the task result cache for the current context.

def set_task_cache(cache: dict[str, Any] | None): (source)

Set the task result cache for the current context.

def task_id(func: Any) -> str: (source)

Return the fully-qualified module.qualname for a function.

Raises ValueError for functions that cannot be identified unambiguously (lambdas, closures, __main__ functions).

_task_cache: ContextVar[dict[str, Any] | None] = (source)

Undocumented