resonate.Resonate

class documentation

class Resonate:

Constructor: Resonate(pid, store, task_source)

View In Hierarchy

The Resonate class serves as the main API interface for Resonate Application Nodes.

Object attributes:
  • _deps (Dependencies): Manages application-level dependencies.
  • _registry (FunctionRegistry): Stores and manages registered functions.
  • _scheduler (IScheduler): Manages coroutine and function executions.
Method __init__ Defaults to a generated UUID.
Method register Undocumented
Method run Undocumented
Method set_dependency Sets a dependency to be used by the Application Node.
Method stop Stops the scheduler and halts all executions.
Instance Variable _deps Undocumented
Instance Variable _registry Undocumented
Instance Variable _scheduler Undocumented
def __init__(self, pid: str | None = None, store: LocalStore | RemoteStore | None = None, task_source: ITaskSource | None = None):

Initialization args:
  • pid (str | None): Optional process ID for the scheduler.
    Defaults to a generated UUID.
  • store (LocalStore | RemoteStore | None): Optional store for promises.
    Defaults to RemoteStore.
  • task_source (TaskSource | None): Optional task source for obtaining tasks.
    Defaults to Poller.

Example:

from resonate import Resonate
# ...
resonate = Resonate()

Example with custom store and task source:

from resonate import Resonate
from resonate.stores import RemoteStore
from resonate.task_sources import Poller
# ...
resonate = Resonate(
    store=RemoteStore(url="http://localhost:8001"),
    task_source=Poller(url="http://localhost:8002"),
)

@overload
def register(self, func: Callable[Concatenate[Context, P], Generator[Yieldable, Any, T]], name: str | None = None, version: int = 1, retry_policy: retry_policy.RetryPolicy | None = None) -> RegisteredFn[P, T]:
@overload
def register(self, func: Callable[Concatenate[Context, P], Coroutine[Any, Any, T]], name: str | None = None, version: int = 1, retry_policy: retry_policy.RetryPolicy | None = None) -> RegisteredFn[P, T]:
@overload
def register(self, func: Callable[Concatenate[Context, P], T], name: str | None = None, version: int = 1, retry_policy: retry_policy.RetryPolicy | None = None) -> RegisteredFn[P, T]:

Undocumented

@overload
def run(self, id: str, func: Callable[Concatenate[Context, P], Generator[Yieldable, Any, T]] | Callable[Concatenate[Context, P], T] | Callable[Concatenate[Context, P], Coroutine[Any, Any, T]], /, *args: P.args, **kwargs: P.kwargs) -> Handle[T]:
@overload
def run(self, id: str, func: tuple[Callable[Concatenate[Context, P], Generator[Yieldable, Any, T]] | Callable[Concatenate[Context, P], T] | Callable[Concatenate[Context, P], Coroutine[Any, Any, T]], _RunOptions], /, *args: P.args, **kwargs: P.kwargs) -> Handle[T]:

Undocumented

def set_dependency(self, key: str, obj: Any):

Sets a dependency to be used by the Application Node.

Parameters
key:strThe identifier for the dependency.
obj:AnyThe dependency object to associate with the key.
def stop(self):

Stops the scheduler and halts all executions.

_deps =

Undocumented

_registry =

Undocumented

_scheduler: IScheduler =

Undocumented