realtime

import asyncio import os from typing import Optional from realtime import AsyncRealtimeClient, RealtimeSubscribeStates async def main(): REALTIME_URL = os.environ.get('REALTIME_URL', 'ws://localhost:4000/websocket') API_KEY = os.environ.get('SUPABASE_ANON_KEY', 'YOUR_SUPABASE_ANON_KEY') # Or your service role key print(f"Connecting to {REALTIME_URL} with API Key: {API_KEY[:5]}...") socket = AsyncRealtimeClient(REALTIME_URL, API_KEY) channel = socket.channel("my_test_channel") def _on_subscribe(status: RealtimeSubscribeStates, err: Optional[Exception]): if status == RealtimeSubscribeStates.SUBSCRIBED: print("Connected to channel!") elif status == RealtimeSubscribeStates.CHANNEL_ERROR: print(f"Error subscribing to channel: {err}") elif status == RealtimeSubscribeStates.TIMED_OUT: print("Subscription timed out.") elif status == RealtimeSubscribeStates.CLOSED: print("Channel unexpectedly closed.") # Listen for database changes (example: all changes in 'public' schema) channel.on_postgres_changes( "*", schema="public", callback=lambda payload: print("Database change received:", payload) ) # Listen for broadcast messages channel.on_broadcast("my-event", lambda payload: print("Broadcast received:", payload)) await channel.subscribe(_on_subscribe) await asyncio.sleep(60) # Keep the client alive for 60 seconds await socket.disconnect() if __name__ == "__main__": # Set these environment variables or replace placeholders for a live connection # os.environ['REALTIME_URL'] = 'wss://<project_ref>.supabase.co/realtime/v1' # os.environ['SUPABASE_ANON_KEY'] = 'eyJ...' asyncio.run(main())