SpacetimeDB SDK for Python

Overview

This repository contains the Python SDK for SpacetimeDB. The SDK allows to interact with the database server and is prepared to work with code generated from a SpacetimeDB backend code.

Documentation

The Python SDK has a Quick Start guide and a Reference.

Installation

The SDK is available as a PyPi package. To install it, run the following command:

pip install spacetimedb-sdk

Usage

The Python SpacetimeDB SDK utilizes a client that uses the asyncio package to provide an event driven interface.

Connecting to SpacetimeDB

To connect to SpacetimeDB, you first need to create a SpacetimeDBAsyncClient instance. The SpacetimeDBAsyncClient constructor takes the module_bindings folder that contains the auto-generated files as a parameter. The module_bindings folder is generated by the CLI generate command.

from spacetimedb_sdk.spacetimedb_async_client import SpacetimeDBAsyncClient

spacetime_client = SpacetimeDBAsyncClient(module_bindings)

To connect to SpacetimeDB, you need to call the run method on the SpacetimeDBAsyncClient instance. The run method takes the following parameters:

• auth_token: The authentication token to use to connect to SpacetimeDB. This token is generated by the backend code and is used to authenticate the client.
• host_name: The hostname of the SpacetimeDB server. This hostname should also contain the port number. For example: http://localhost:3000.
• module_address: The address of the module to connect to. This is the same address that you use to connect to the SpacetimeDB web interface.
• ssl_enabled: Whether to use SSL to connect to SpacetimeDB. True if connecting to SpacetimeDB Cloud.
• on_connect: A callback that is called when the client connects to SpacetimeDB.
• queries: A list of queries to subscribe to. The queries are the same queries that you use to subscribe to tables in the SpacetimeDB web interface.

Example:

import asyncio

asyncio.run(
    spacetime_client.run(
        local_config.get_string("auth_token"),
        "http://localhost:3000",
        "chatqs",
        on_connect,
        ["SELECT * FROM User", "SELECT * FROM Message"],
    )
)

Listening to events

To listen to events, you need to register callbacks on the SpacetimeDBAsyncClient instance. The following callbacks are available:

• on_subscription_applied: Called when the client receives the initial data from SpacetimeDB after subscribing to tables.
• scheduled events: You can schedule events to be called at a later time. The schedule_event method takes the following parameters:
  • delay: The delay in seconds before the event is called.
  • callback: The callback to call when the event is called.

You can register for row update events on a table. To do this, you need to register callbacks on the auto-generated table class. The following callbacks are available:

• register_row_update: Called when a row is inserted, updated, or deleted from the table. The callback takes the following parameters:
  • row_op: The operation that was performed on the row. Can be insert, update, or delete.
  • row_old: The old row value. None if the operation is insert.
  • row: The new row value. None if the operation is delete.
  • reducer_event: The reducer event that caused the row update. None if the row update was not caused by a reducer event.

Example:

def on_message_row_update(row_op, message_old, message, reducer_event):
    if reducer_event is not None and row_op == "insert":
        print_message(message)

Message.register_row_update(on_message_row_update)

You can register for reducer call updates as well.

• register_on_REDUCER: Called when a reducer call is received from SpacetimeDB. (If a) you are subscribed to the table that the reducer modifies or b) You called the reducer and it failed)

Example:

def on_send_message_reducer(sender, status, message, msg):
    if sender == local_identity:
        if status == "failed":
            print(f"Failed to send message: {message}")

send_message_reducer.register_on_send_message(on_send_message_reducer)

Accessing the client cache

The client cache is a local cache of the data that the client has received from SpacetimeDB. The client cache is automatically updated when the client receives updates from SpacetimeDB.

When you run the CLI generate command, SpacetimeDB will automatically generate a class for each table in your database.

• filter_by_COLUMN: Filters the table by the specified column value.
• iter: Returns an iterator over the table.

Example:

from module_bindings.user import User

my_user = User.filter_by_identity(local_identity)

for user in User.iter():
    print(user.name)

Calling Reducers

To call a reducer, you need to call the autogenerated method in the auto-generated reducer file.

Example:

import module_bindings.send_message_reducer as send_message_reducer

send_message_reducer.send_message("Hello World!")