Resonate Python SDK

Welcome to the Resonate Python SDK guide! This SDK makes it possible to write Distributed Async Await applications with Python. This guide covers installation and features that the SDK offers.

➡️ Resonate Python SDK API reference

Installation

How to install the Resonate Python SDK into your project.

To install the Resonate Python SDK, you can use any of your favorite package managers.

Rye

rye add resonate-sdk

Pip

pip install resonate-sdk

Then initialize the SDK in your project, register your top-level function(s) and call .run():

from resonate.scheduler import Scheduler

# ...

resonate = Scheduler(durable_promise_storage=LocalPromiseStore())
resonate.register(your_func)
resonate.run(your_func)

Promise storage modes

The Resonate Python SDK supports two storage modes: Local and Remote.

Local storage

How to use Local storage mode in the Resonate Python SDK.

Local storage mode is ideal for starting out with Resonate.

resonate = Scheduler(durable_promise_storage=LocalPromiseStore())

Remote storage

How to use Remote storage mode in the Resonate Python SDK.

Remote storage ensures promises are stored in the Resonate Server. To enable Remote storage, pass the Resonate Server's address when initializing Resonate:

resonate = Scheduler(durable_promise_storage=RemotePromiseStore(url="http://localhost:8001"))

Set a Retry Policy

How to set a Retry Policy in the Resonate Python SDK.

You can set a Retry Policy to control how a function should be retried if it fails. A Retry Policy can be set at the function level or when the individual function is invoked.

There are several different built in Retry Policies available, and each one can be refined to your needs.

Function registration

Here is an example of setting an Exponential Retry Policy when registering a function:

from resonate.retry_policy import Exponential

// ...

resonate.register(your_func, retry_policy=Exponential())

Function invocation

Here is an example of setting a Constant Retry Policy when invoking a function:

from resonate.retry_policy import Constant

// ...

ctx.lfc(another_func, retry_policy=Constant())

Dependency injection

How to use Dependency injection in the Resonate Python SDK.

You can inject dependencies into your Resonate instance and use them in functions that receive the Resonate Context.

from resonate.scheduler import Scheduler
from resonate.storage import LocalPromiseStore

# ...

s = Scheduler(durable_promise_storage=LocalPromiseStore())
# ...
s._deps.set("dependency-a", dependency())
s._deps.set("dependency-b", "some-dependency")

Then you can access the dependencies in your functions:

def your_func(ctx: Context):
# ...
dep = ctx.deps.get("dependency-a")

Batch operations

Resonate's transparent batching feature handles the coordination of otherwise concurrent executions to create batches, enabling you to write concurrent, non-coordinated code. For a deeper dive into transparent batching, check out the Transparent batching with the Resonate Python SDK blog post. To use transparent batching, follow these steps.

First, create a data structure that inherits what Resonate calls a Command interface. The data structure must include the data to be inserted into the database. The Command data structure stands in for a function execution invocation so that you still get a Durable Promise and await on result.

from dataclasses import dataclass
# ...
from resonate.commands import Command
# ...
# Define a data structure for the Resonate SDK to track and create batches of
@dataclass
class InsertUser(Command):
    id: int

Then, create a handler that can process a batch of operations.

# ...
from resonate.context import Context
# ...
# Define a function that inserts a batch of rows into the database
# The main difference is that commit() is only called after all the Insert statements are executed
def _batch_handler(_: Context, users: list[InsertUser]):
    # error handling ommitted for this example
    for user in users:
        conn.execute("INSERT INTO users (value) VALUES (?)", (user.id,))
    conn.commit()
    print(f"{len(users)} users have been inserted to database.")

Next, register the data structure and the handler with the Resonate Scheduler.

# ...
from resonate.scheduler import Scheduler
from resonate.storage import LocalPromiseStore
from resonate.retry_policy import never
# ...
# Create a Resonate Scheduler
resonate = Scheduler(LocalPromiseStore(), processor_threads=1)
# ...
# Register the batch handler and data structure with the Resonate Scheduler
resonate.register_command_handler(InsertUser, _batch_handler, retry_policy=never())

Finally, create a function that can be invoked over and over again and passes the data to Resonate to manage. Register it with the Resonate Scheduler, and then call that function with Resonate's run() method.

# ...
# Definte the top level function that uses batching
def create_user_batching(ctx: Context, u: int):
    p = yield ctx.lfi(InsertUser(u))
    yield p
# ...
# Register the top level functions with the Resonate Scheduler
resonate.register(create_user_batching, retry_policy=never())
# ...
def main() -> None:
# ...
    # Create an array to hold the promises
    promises = []

    for u in range(10000):
        p = resonate.run(f"insert-value-{u}", create_user_batching, u)
        promises.append(p)

    for p in promises:
        p.result()