socketIO-client 0.7.1 on PyPI

socketIO-client

Here is a socket.io client library for Python. You can use it to write test code for your socket.io server.

Please note that this version implements socket.io protocol 1.x, which is not backwards compatible. If you want to communicate using socket.io protocol 0.9 (which is compatible with gevent-socketio), please use socketIO-client 0.5.7.2.

Installation

Install the package in an isolated environment.

VIRTUAL_ENV=$HOME/.virtualenv

# Prepare isolated environment
virtualenv $VIRTUAL_ENV

# Activate isolated environment
source $VIRTUAL_ENV/bin/activate

# Install package
pip install -U socketIO-client

Usage

Activate isolated environment.

VIRTUAL_ENV=$HOME/.virtualenv
source $VIRTUAL_ENV/bin/activate

Launch your socket.io server.

cd $(python -c "import os, socketIO_client;\
    print(os.path.dirname(socketIO_client.__file__))")

DEBUG=* node tests/serve.js  # Start socket.io server in terminal one
DEBUG=* node tests/proxy.js  # Start proxy server in terminal two
nosetests                    # Run tests in terminal three

For debugging information, run these commands first.

import logging
logging.getLogger('socketIO-client').setLevel(logging.DEBUG)
logging.basicConfig()

Emit.

from socketIO_client import SocketIO, LoggingNamespace

with SocketIO('localhost', 8000, LoggingNamespace) as socketIO:
    socketIO.emit('aaa')
    socketIO.wait(seconds=1)

Emit with callback.

from socketIO_client import SocketIO, LoggingNamespace

def on_bbb_response(*args):
    print('on_bbb_response', args)

with SocketIO('localhost', 8000, LoggingNamespace) as socketIO:
    socketIO.emit('bbb', {'xxx': 'yyy'}, on_bbb_response)
    socketIO.wait_for_callbacks(seconds=1)

Define events.

from socketIO_client import SocketIO, LoggingNamespace

def on_connect():
    print('connect')

def on_disconnect():
    print('disconnect')

def on_reconnect():
    print('reconnect')

def on_aaa_response(*args):
    print('on_aaa_response', args)

socketIO = SocketIO('localhost', 8000, LoggingNamespace)
socketIO.on('connect', on_connect)
socketIO.on('disconnect', on_disconnect)
socketIO.on('reconnect', on_reconnect)

# Listen
socketIO.on('aaa_response', on_aaa_response)
socketIO.emit('aaa')
socketIO.emit('aaa')
socketIO.wait(seconds=1)

# Stop listening
socketIO.off('aaa_response')
socketIO.emit('aaa')
socketIO.wait(seconds=1)

# Listen only once
socketIO.once('aaa_response', on_aaa_response)
socketIO.emit('aaa')  # Activate aaa_response
socketIO.emit('aaa')  # Ignore
socketIO.wait(seconds=1)

Define events in a namespace.

from socketIO_client import SocketIO, BaseNamespace

class Namespace(BaseNamespace):

    def on_aaa_response(self, *args):
        print('on_aaa_response', args)
        self.emit('bbb')

socketIO = SocketIO('localhost', 8000, Namespace)
socketIO.emit('aaa')
socketIO.wait(seconds=1)

Define standard events.

from socketIO_client import SocketIO, BaseNamespace

class Namespace(BaseNamespace):

    def on_connect(self):
        print('[Connected]')

    def on_reconnect(self):
        print('[Reconnected]')

    def on_disconnect(self):
        print('[Disconnected]')

socketIO = SocketIO('localhost', 8000, Namespace)
socketIO.wait(seconds=1)

Define different namespaces on a single socket.

from socketIO_client import SocketIO, BaseNamespace

class ChatNamespace(BaseNamespace):

    def on_aaa_response(self, *args):
        print('on_aaa_response', args)

class NewsNamespace(BaseNamespace):

    def on_aaa_response(self, *args):
        print('on_aaa_response', args)

socketIO = SocketIO('localhost', 8000)
chat_namespace = socketIO.define(ChatNamespace, '/chat')
news_namespace = socketIO.define(NewsNamespace, '/news')

chat_namespace.emit('aaa')
news_namespace.emit('aaa')
socketIO.wait(seconds=1)

Connect via SSL (https://github.com/invisibleroads/socketIO-client/issues/54).

from socketIO_client import SocketIO

# Skip server certificate verification
SocketIO('https://localhost', verify=False)
# Verify the server certificate
SocketIO('https://localhost', verify='server.crt')
# Verify the server certificate and encrypt using client certificate
socketIO = SocketIO('https://localhost', verify='server.crt', cert=(
    'client.crt', 'client.key'))

Specify params, headers, cookies, proxies thanks to the requests library.

from socketIO_client import SocketIO
from base64 import b64encode

SocketIO(
    'localhost', 8000,
    params={'q': 'qqq'},
    headers={'Authorization': 'Basic ' + b64encode('username:password')},
    cookies={'a': 'aaa'},
    proxies={'https': 'https://proxy.example.com:8080'})

Wait forever.

from socketIO_client import SocketIO

socketIO = SocketIO('localhost', 8000)
socketIO.wait()

Don't wait forever.

from requests.exceptions import ConnectionError
from socketIO_client import SocketIO

try:
    socket = SocketIO('localhost', 8000, wait_for_connection=False)
    socket.wait()
except ConnectionError:
    print('The server is down. Try again later.')