resonate.Context

class documentation

class Context:

Constructor: Context(id, cid, info, registry, ...)

Undocumented

Method	`__init__`	Undocumented
Method	`__repr__`	Undocumented
Method	`begin_rpc`	Schedules a function for remote execution and returns a promise.
Method	`begin_run`	Schedules a function for an immediate effectively-once local execution and returns a promise.
Method	`detached`	Schedules a function for immediate effectively-once execution that is completely detached from the current Call Graph, starting a new Call Graph.
Method	`get_dependency`	Retrieve a registered dependency by name.
Method	`lfc`	Undocumented
Method	`lfi`	Undocumented
Method	`promise`	Get or creates a durable promise that can be awaited.
Method	`rfc`	Undocumented
Method	`rfi`	Undocumented
Method	`rpc`	Schedules a function for remote execution and awaits its result.
Method	`run`	Schedules a function for an immediate effectively-once local execution and awaits its result.
Method	`sleep`	Suspends execution within a Resonate function for a specified duration.
Method	`typesafe`	Optionally provide type safety for your coroutine definition.
Property	`id`	Id for the current execution.
Property	`info`	Information for the current execution.
Property	`logger`	Undocumented
Property	`random`	Deterministic random methods.
Property	`time`	Deterministic time methods.
Method	`_next`	Undocumented
Instance Variable	`_cid`	Undocumented
Instance Variable	`_counter`	Undocumented
Instance Variable	`_dependencies`	Undocumented
Instance Variable	`_id`	Undocumented
Instance Variable	`_info`	Undocumented
Instance Variable	`_logger`	Undocumented
Instance Variable	`_random`	Undocumented
Instance Variable	`_registry`	Undocumented
Instance Variable	`_time`	Undocumented

def __init__(self, id: str, cid: str, info: Info, registry: Registry, dependencies: Dependencies, logger: Logger): ¶

Undocumented

def __repr__(self) -> str: ¶

Undocumented

@overload

def begin_rpc(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> RFI[R]:

@overload

def begin_rpc(self, func: str, *args: Any, **kwargs: Any) -> RFI:

Schedules a function for remote execution and returns a promise.

This method schedules the given function for remote execution through the global event loop. The function may run in a different process or worker, depending on the configured Resonate environment. It serves as an alias for ctx.rfi, providing a simplified interface for non-blocking remote calls.

The returned promise can be awaited to obtain the final result once the remote execution completes.

The function must be registered with Resonate before invocation, and all arguments must be serializable.

Parameters
func:`Callable \| str`	The function to execute remotely, or the registered name of the function.
*args:`Any`	Positional arguments to pass to the function.
**kwargs:`Any`	Keyword arguments to pass to the function.

Returns
`RFI`	An object representing the scheduled execution, which can be awaited for the final result.

def begin_run(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> LFI[R]: ¶

Schedules a function for an immediate effectively-once local execution and returns a promise.

This method initiates the immediate asynchronous execution of the given function within the current process, returning a promise — yield the promise to get a result. This method checkpoints at the invocation and result of the function execution. This method is an alias of ctx.lfi().

By default, execution is durable, but non-durable behavior can be configured if desired.

Parameters
func:`Callable[Concatenate[Context, P], Generator[Any, Any, R] \| R]`	The function to execute locally.
*args:`P.args`	Positional arguments to pass to the function.
**kwargs:`P.kwargs`	Keyword arguments to pass to the function.

Returns
`LFI[R]`	An object representing the scheduled local execution, which can be awaited for the final result.

@overload

def detached(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> RFI[R]:

@overload

def detached(self, func: str, *args: Any, **kwargs: Any) -> RFI:

Schedules a function for immediate effectively-once execution that is completely detached from the current Call Graph, starting a new Call Graph.

This method schedules the given function for remote execution through the global event loop. The function may run in a separate process or worker without being tied to the callers lifecycle. It is typically used for background or fire-and-forget workloads.

The returned promise can still be awaited if the caller wishes to retrieve the final result upon completion.

The function must be registered with Resonate before invocation, and all arguments must be serializable.

Parameters
func:`Callable \| str`	The function to execute remotely, or the registered name of the function.
*args:`Any`	Positional arguments to pass to the function.
**kwargs:`Any`	Keyword arguments to pass to the function.

Returns
`RFI`	A remote function invocation handle representing the detached execution, which can optionally be awaited for the final result.

def get_dependency(self, key: str, default: T = None) -> Any | T: ¶

Retrieve a registered dependency by name.

This method returns the dependency object registered under the given key. If no dependency exists for that key, the provided default value is returned. A TypeError is raised if key is not a string.

Example

Retrieving dependencies from a context:

db = ctx.get_dependency("db")
cache = ctx.get_dependency("cache", default=NullCache())

Parameters
key:`str`	The name of the dependency to retrieve.
default:`T, optional`	The value to return if no dependency is found. Defaults to `None`.

Returns
`Any \| T`	The registered dependency if it exists, otherwise the specified `default` value.

Raises
`TypeError`	If `key` is not a string.

def lfc(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> LFC[R]: ¶

Undocumented

def lfi(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> LFI[R]: ¶

Undocumented

def promise(self, *, id: str | None = None, timeout: float | None = None, idempotency_key: str | None = None, data: Any = None, tags: dict[str, str] | None = None) -> RFI: ¶

Get or creates a durable promise that can be awaited.

If no id is provided, a new promise is created with a generated identifier. If an id is provided and a promise with that ID already exists, the existing promise is returned (provided the idempotency_key matches). This ensures safe, idempotent reuse of promises across invocations or retries.

Promises are especially useful for Human-In-The-Loop (HITL) workflows, where execution needs to pause until a human provides input or takes an action. The created promise can later be fulfilled using Resonate.promise.resolve().

Parameters
id:`str, optional`	The unique identifier of the promise. If omitted, a new ID is generated automatically.
timeout:`float, optional`	The duration in seconds before the promise expires. If `None`, the promise does not expire.
idempotency_key:`str, optional`	A key used to ensure idempotent creation. When specified, an existing promise with the same key will be reused.
data:`Any, optional`	Optional initial data to associate with the promise.
tags:`dict[str, str], optional`	Metadata tags for categorizing or tracking the promise.

Returns
`RFI`	An object representing the created or retrieved promise. The handle can be awaited to block until the promise is resolved.

@overload

def rfc(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> RFC[R]:

@overload

def rfc(self, func: str, *args: Any, **kwargs: Any) -> RFC:

Undocumented

@overload

def rfi(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> RFI[R]:

@overload

def rfi(self, func: str, *args: Any, **kwargs: Any) -> RFI:

Undocumented

@overload

def rpc(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> RFC[R]:

@overload

def rpc(self, func: str, *args: Any, **kwargs: Any) -> RFC:

Schedules a function for remote execution and awaits its result.

This method executes the given function remotely through the global event loop. The function may run in a different process or worker, depending on the configured Resonate environment. It serves as an alias for ctx.rfc, providing a simplified interface for blocking remote calls.

The function must be registered with Resonate before invocation, and all arguments must be serializable.

Parameters
func:`Callable \| str`	The function to execute remotely, or the registered name of the function.
*args:`Any`	Positional arguments to pass to the function.
**kwargs:`Any`	Keyword arguments to pass to the function.

Returns
`RFC`	An object representing the execution and its result.

def run(self, func: Callable[Concatenate[Context, P], Generator[Any, Any, R] | R], *args: P.args, **kwargs: P.kwargs) -> LFC[R]: ¶

Schedules a function for an immediate effectively-once local execution and awaits its result.

This method initiates the immediate effectively-once execution of the given function within the current process. By default, this method checkpoints at the invocation and the result of the called function. This method is an alias of ctx.lfc().

Parameters
func:`Callable[Concatenate[Context, P], Generator[Any, Any, R] \| R]`	The function to execute locally.
*args:`P.args`	Positional arguments to pass to the function.
**kwargs:`P.kwargs`	Keyword arguments to pass to the function.

Returns
`LFC[R]`	An object representing the scheduled execution and its eventual result.

def sleep(self, secs: float) -> RFC[None]: ¶

Suspends execution within a Resonate function for a specified duration.

This method pauses the current function for the given number of seconds. It can be safely used within both local and remote Resonate-managed functions to simulate delays, throttle execution, or coordinate timing between tasks.

There is no maximum sleep duration. Fractional (floating-point) values are supported for sub-second precision.

Parameters
secs:`float`	The number of seconds to sleep. Can be a fractional value.

Returns
`RFC[None]`	A handle representing the sleep operation, which resolves when the specified duration has elapsed.

@overload

def typesafe(self, cmd: LFI[T] | RFI[T]) -> Generator[LFI[T] | RFI[T], Promise[T], Promise[T]]:

@overload

def typesafe(self, cmd: LFC[T] | RFC[T] | Promise[T]) -> Generator[LFC[T] | RFC[T] | Promise[T], T, T]:

Optionally provide type safety for your coroutine definition.

@property

id: str = ¶

Id for the current execution.

@property

info: Info = ¶

Information for the current execution.

@property

logger: Logger = ¶

Undocumented

@property

random: Random = ¶

Deterministic random methods.

@property

time: Time = ¶

Deterministic time methods.

def _next(self) -> str: ¶

Undocumented

_cid = ¶

Undocumented

_counter: int = ¶

Undocumented

_dependencies = ¶

Undocumented

_id = ¶

Undocumented

_info = ¶

Undocumented

_logger = ¶

Undocumented

_random = ¶

Undocumented

_registry = ¶

Undocumented

_time = ¶

Undocumented