Supabase Realtime Python Client

Broadcast allows sending low-latency, ephemeral messages between clients.

Configuration Options:

• ack (bool): If true, the send_broadcast call will resolve once the server acknowledges receipt.
• self (bool): If true, the client that sent the message will also receive it.
• private (bool): If true, Row Level Security (RLS) is used to determine connection permissions.

API Methods:

• channel.on_broadcast(event, callback): Register a listener for a specific event name.
• channel.send_broadcast(event, payload): Send a message with a specific event name and JSON payload.

# Example setup
channel = client.channel(
    "broadcast-test", {"config": {"broadcast": {"ack": False, "self": False}}}
)

# Listen for events
await channel.on_broadcast("some-event", lambda payload: print(payload)).subscribe()

# Send events
await channel.send_broadcast("some-event", {"hello": "world"})

Presence allows clients to track and synchronize shared state (like online users) using CRDTs.

Configuration:

• Use the presence key in the channel config to specify a key.

API Methods:

• channel.on_presence_sync(callback): Triggered when the local presence state is synchronized.
• channel.on_presence_join(callback): Triggered when new users join the channel.
• channel.on_presence_leave(callback): Triggered when users leave the channel.
• channel.track(payload): Updates the current client's presence state.
• channel.presence_state(): Returns the current synchronized state of all participants.

channel = client.channel(
    "presence-test",
    {
        "config": {
            "presence": {
                "key": ""
            }
        }
    }
)

channel.on_presence_sync(lambda: print("Online users: ", channel.presence_state()))
channel.on_presence_join(lambda new_presences: print("New users have joined: ", new_presences))
channel.on_presence_leave(lambda left_presences: print("Users have left: ", left_presences))

await channel.track({ 'user_id': 1 })

To use Realtime, instantiate an AsyncRealtimeClient with your REALTIME_URL and API_KEY, then create a channel using socket.channel("channel-name"). You can subscribe to a channel and provide a callback to handle subscription status updates using RealtimeSubscribeStates.

import asyncio
from typing import Optional
from realtime import AsyncRealtimeClient, RealtimeSubscribeStates

async def main():
    # Use ws://localhost:4000/websocket for local dev
    # Use wss://<project_ref>.supabase.co/realtime/v1 for Supabase projects
    REALTIME_URL = "ws://localhost:4000/websocket"
    API_KEY = "1234567890"

    socket = AsyncRealtimeClient(REALTIME_URL, API_KEY)
    channel = socket.channel("test-channel")

    def _on_subscribe(status: RealtimeSubscribeStates, err: Optional[Exception]):
        if status == RealtimeSubscribeStates.SUBSCRIBED:
            print("Connected!")
        elif status == RealtimeSubscribeStates.CHANNEL_ERROR:
            print(f"There was an error subscribing to channel: {err.args}")
        elif status == RealtimeSubscribeStates.TIMED_OUT:
            print("Realtime server did not respond in time.")
        elif status == RealtimeSubscribeStates.CLOSED:
            print("Realtime channel was unexpectedly closed.")

    await channel.subscribe(_on_subscribe)

Postgres CDC allows you to listen for changes (INSERT, UPDATE, DELETE, etc.) in your database tables.

API Method:

• channel.on_postgres_changes(event, schema, table, filter, callback)

Parameters:

• event: The type of change to listen for (e.g., "INSERT", "UPDATE", or "*" for all).
• schema: The database schema (e.g., "public").
• table: The specific table to monitor.
• filter: A string filter for specific rows (e.g., "username=eq.Realtime").
• callback: A function that receives the change payload.

channel = client.channel("db-changes")

# Listen to all changes in public schema
channel.on_postgres_changes(
    "*",
    schema="public",
    callback=lambda payload: print("All changes in public schema: ", payload),
)

# Listen to specific table inserts
channel.on_postgres_changes(
    "INSERT",
    schema="public",
    table="messages",
    callback=lambda payload: print("All inserts in messages table: ", payload),
)

# Listen to specific updates with a filter
channel.on_postgres_changes(
    "UPDATE",
    schema="public",
    table="users",
    filter="username=eq.Realtime",
    callback=lambda payload: print(
        "All updates on users table when username is Realtime: ", payload
    ),
)

await channel.subscribe(
    lambda status, err: status == RealtimeSubscribeStates.SUBSCRIBED
    and print("Ready to receive database changes!")
)