realtimex-python 1.0.4

Python SDK for RealtimeX real-time messaging service

RealtimeX Python SDK

Python client library for RealtimeX real-time messaging service.

Installation

pip

pip install realtimex-python

Quick Start

from realtimex import RealtimeX

# Initialize
realtimex = RealtimeX('YOUR_API_KEY', {
    'cluster': 'eu',  # optional, default 'eu'
    'ws_host': 'ws.realtimex.net',  # optional
    'ws_port': 443,  # optional
    'encrypted': True,  # optional, default True
})

# Bind connection events
realtimex.connection.bind('connected', lambda: print('Connected!'))

# Connect to server
realtimex.connect()

# Subscribe to a channel
channel = realtimex.subscribe('my-channel')

# Listen for events
channel.bind('my-event', lambda data: print('Received:', data))

# Send client events
channel.trigger('client-my-event', {
    'message': 'Hello'
})

API

RealtimeX(api_key, options={})

Create a new RealtimeX instance.

Options:

• cluster (str): Cluster name, default 'eu'
• ws_host (str): WebSocket host, default 'ws.realtimex.net'
• ws_port (int): WebSocket port, default 443
• encrypted (bool): Use WSS, default True

Methods

subscribe(channel_name)

Subscribe to a channel.

channel = realtimex.subscribe('my-channel')

unsubscribe(channel_name)

Unsubscribe from a channel.

realtimex.unsubscribe('my-channel')

disconnect()

Disconnect from RealtimeX.

realtimex.disconnect()

Channel

bind(event, callback)

Bind to an event.

channel.bind('my-event', lambda data: print(data))

unbind(event, callback=None)

Unbind from an event.

channel.unbind('my-event')

trigger(event, data)

Trigger a client event (must be prefixed with client-).

channel.trigger('client-my-event', {'message': 'Hello'})

bind_global(callback)

Bind to all events on the channel.

channel.bind_global(lambda event, data: print(event, data))

Connection Events

realtimex.connection.bind('connecting', lambda: print('Connecting...'))
realtimex.connection.bind('connected', lambda: print('Connected!'))
realtimex.connection.bind('disconnected', lambda: print('Disconnected'))

Private Channels

Option 1: With auth_endpoint (automatic):

realtimex = RealtimeX('YOUR_API_KEY', {
    'ws_host': 'localhost',
    'ws_port': 3001,
    'encrypted': False,
    'auth_endpoint': 'http://localhost:3001/auth'  # SDK makes HTTP request automatically
})

realtimex.connect()
channel = realtimex.subscribe('private-my-channel')  # SDK handles auth automatically

Option 2: With custom auth_callback:

import requests

def authorizer(channel_name, socket_id):
    # Custom auth logic
    response = requests.post('http://localhost:3001/auth/channels', json={
        'channel_name': channel_name,
        'socket_id': socket_id
    })
    return response.json()  # Returns: {'auth': 'real_token', 'channel_data': '...'}

realtimex = RealtimeX('YOUR_API_KEY', 
    options={'ws_host': 'localhost', 'ws_port': 3001, 'encrypted': False},
    auth_callback=authorizer
)

realtimex.connect()
channel = realtimex.subscribe('private-my-channel')

Presence channels with user data:

realtimex = RealtimeX('YOUR_API_KEY',
    options={
        'ws_host': 'localhost',
        'ws_port': 3001,
        'encrypted': False,
        'auth_endpoint': 'http://localhost:3001/auth'
    },
    user_data={'user_id': '123', 'name': 'John'}
)

realtimex.connect()
channel = realtimex.subscribe('presence-room-1')

License

MIT