{"library":"realtime","code":"import asyncio\nimport os\nfrom typing import Optional\nfrom realtime import AsyncRealtimeClient, RealtimeSubscribeStates\n\nasync def main():\n    REALTIME_URL = os.environ.get('REALTIME_URL', 'ws://localhost:4000/websocket')\n    API_KEY = os.environ.get('SUPABASE_ANON_KEY', 'YOUR_SUPABASE_ANON_KEY') # Or your service role key\n\n    print(f\"Connecting to {REALTIME_URL} with API Key: {API_KEY[:5]}...\")\n\n    socket = AsyncRealtimeClient(REALTIME_URL, API_KEY)\n    channel = socket.channel(\"my_test_channel\")\n\n    def _on_subscribe(status: RealtimeSubscribeStates, err: Optional[Exception]):\n        if status == RealtimeSubscribeStates.SUBSCRIBED:\n            print(\"Connected to channel!\")\n        elif status == RealtimeSubscribeStates.CHANNEL_ERROR:\n            print(f\"Error subscribing to channel: {err}\")\n        elif status == RealtimeSubscribeStates.TIMED_OUT:\n            print(\"Subscription timed out.\")\n        elif status == RealtimeSubscribeStates.CLOSED:\n            print(\"Channel unexpectedly closed.\")\n\n    # Listen for database changes (example: all changes in 'public' schema)\n    channel.on_postgres_changes(\n        \"*\",\n        schema=\"public\",\n        callback=lambda payload: print(\"Database change received:\", payload)\n    )\n\n    # Listen for broadcast messages\n    channel.on_broadcast(\"my-event\", lambda payload: print(\"Broadcast received:\", payload))\n\n    await channel.subscribe(_on_subscribe)\n    await asyncio.sleep(60) # Keep the client alive for 60 seconds\n    await socket.disconnect()\n\nif __name__ == \"__main__\":\n    # Set these environment variables or replace placeholders for a live connection\n    # os.environ['REALTIME_URL'] = 'wss://<project_ref>.supabase.co/realtime/v1'\n    # os.environ['SUPABASE_ANON_KEY'] = 'eyJ...'\n    asyncio.run(main())","lang":"python","description":"This quickstart demonstrates how to establish a connection to the Supabase Realtime server, subscribe to a channel, listen for PostgreSQL database changes, and receive broadcast messages. Remember to replace placeholder `REALTIME_URL` and `SUPABASE_ANON_KEY` with your actual Supabase project details, typically retrieved from your project settings. Ensure your API key has the necessary permissions (e.g., 'anon' key for public read access or a 'service_role' key for elevated privileges).","tag":null,"tag_description":null,"last_tested":"2026-04-24","results":[{"runtime":"python:3.10-alpine","exit_code":1},{"runtime":"python:3.10-slim","exit_code":1},{"runtime":"python:3.11-alpine","exit_code":1},{"runtime":"python:3.11-slim","exit_code":1},{"runtime":"python:3.12-alpine","exit_code":1},{"runtime":"python:3.12-slim","exit_code":1},{"runtime":"python:3.13-alpine","exit_code":1},{"runtime":"python:3.13-slim","exit_code":1},{"runtime":"python:3.9-alpine","exit_code":1},{"runtime":"python:3.9-slim","exit_code":1}]}