socketIO-client

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

Installation

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

For debugging information, run these commands first.

import logging
logging.basicConfig(level=logging.DEBUG)

Emit.

from socketIO_client import SocketIO

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

Emit with callback.

from socketIO_client import SocketIO

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

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

Define events.

from socketIO_client import SocketIO

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

socketIO = SocketIO('localhost', 8000)
socketIO.on('aaa_response', on_aaa_response)
socketIO.emit('aaa')
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]'

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.

from socketIO_client import SocketIO

SocketIO('https://localhost')

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')
socketIO.wait()

License

This software is available under the MIT License.

Credits

• Guillermo Rauch wrote the socket.io specification.
• Hiroki Ohtani wrote websocket-client.
• rod wrote a prototype for a Python client to a socket.io server on StackOverflow.
• Alexandre Bourget wrote gevent-socketio, which is a socket.io server written in Python.
• Paul Kienzle, Zac Lee, Josh VanderLinden, Ian Fitzpatrick, Lucas Klein, Rui Chicoria, Travis Odom submitted code to expand support of the socket.io protocol.
• Bernard Pratz and Francis Bull wrote prototypes to support xhr-polling and jsonp-polling.
• Eric Chen, Denis Zinevich, Thiago Hersan, Nayef Copty suggested ways to make the connection more robust.