README.md

Supabase Realtime Client

Send ephemeral messages with Broadcast, track and synchronize state with Presence, and listen to database changes with Postgres Change Data Capture (CDC).

Guides · Reference Docs · Multiplayer Demo

Overview

This client enables you to use the following Supabase Realtime's features:

• Broadcast: send ephemeral messages from client to clients with minimal latency. Use cases include sharing cursor positions between users.
• Presence: track and synchronize shared state across clients with the help of CRDTs. Use cases include tracking which users are currently viewing a specific webpage.
• Postgres Change Data Capture (CDC): listen for changes in your PostgreSQL database and send them to clients.

Usage

Installing the Package

pip3 install realtime

or using uv

uv add realtime

Creating a Channel

import asyncio
from typing import Optional

from realtime import AsyncRealtimeClient, RealtimeSubscribeStates


async def main():
    REALTIME_URL = "ws://localhost:4000/websocket"
    API_KEY = "1234567890"

    client = AsyncRealtimeClient(REALTIME_URL, API_KEY)
    channel = client.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)

Notes:

• REALTIME_URL is ws://localhost:4000/socket when developing locally and wss://<project_ref>.supabase.co/realtime/v1 when connecting to your Supabase project.
• API_KEY is a JWT whose claims must contain exp and role (existing database role).
• Channel name can be any string.

Broadcast

Your client can send and receive messages based on the event.

# Setup...

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

await channel.on_broadcast("some-event", lambda payload: print(payload)).subscribe()
await channel.send_broadcast("some-event", {"hello": "world"})

Notes:

• Setting ack to true means that the channel.send promise will resolve once server replies with acknowledgement that it received the broadcast message request.
• Setting self to true means that the client will receive the broadcast message it sent out.
• Setting private to true means that the client will use RLS to determine if the user can connect or not to a given channel.
• Setting replication_ready to true instructs the server to emit a system event once the Postgres replication connection backing the channel is established and ready to stream changes. Listen for it with on_system; the payload's status is "ok" (message: "Replication connection established") on success or "error" if the connection is not ready in time.

channel = client.channel(
    "db-changes", {"config": {"broadcast": {"replication_ready": True}}}
)

channel.on_postgres_changes(
    "*", schema="public", table="messages",
    callback=lambda payload: print("Change received!", payload),
).on_system(
    lambda payload: payload.status == "ok"
    and print("Replication connection is ready:", payload.message)
)

await channel.subscribe()

Presence

Your client can track and sync state that's stored in the channel.

# Setup...

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 })

Postgres CDC

Receive database changes on the client.

# Setup...

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

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

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

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
    ),
)

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

Filters

filter is a column=operator.value expression evaluated server-side. The following operators are supported:

Operator	Example	Meaning
eq	id=eq.1	equal
neq	id=neq.1	not equal
lt lte gt gte	age=gte.18	comparison
in	status=in.(active,pending)	in list
like ilike	title=like.%foo%	pattern match (case in/sensitive)
is	deleted_at=is.null	IS null/true/false/unknown
match imatch	title=match.^foo	POSIX regex match (~ / ~*)
isdistinct	value=isdistinct.1	NULL-safe inequality

Any operator can be negated with the not. prefix, e.g. filter="status=not.in.(draft,archived)". Multiple conditions combined with commas are applied as an AND, e.g. filter="amount=gt.100,status=in.(open,pending)".

Note: Realtime evaluates filters server-side over a single table's WAL — there is no resource embedding or or() grouping, and % (not *) is the wildcard for like/ilike.

Selecting columns

Use select to receive only a subset of columns instead of the full row. This reduces payload size (helpful for large bytea/jsonb columns). The selected columns must be selectable by the subscribing role:

channel.on_postgres_changes(
    "*",
    schema="public",
    table="users",
    select=["id", "first_name"],
    # payload record only contains { "id": ..., "first_name": ... }
    callback=lambda payload: print("Selected columns only: ", payload),
)

Get All Channels

You can see all the channels that your client has instantiated.

# Setup...

client.get_channels()

Cleanup

It is highly recommended that you clean up your channels after you're done with them.

• Remove a single channel

# Setup...

channel = client.channel('some-channel-to-remove')

channel.subscribe()

await client.remove_channel(channel)

• Remove all channels

# Setup...

channel1 = client.channel('a-channel-to-remove')
channel2 = client.channel('another-channel-to-remove')

await channel1.subscribe()
await channel2.subscribe()

await client.remove_all_channels()

Credits

This repo draws heavily from phoenix-js.

License

MIT.