OddSockets Python SDK Documentation

Overview & Features

The OddSockets Python SDK provides a powerful, easy-to-use interface for real-time messaging in Python applications with full async/await support.

Python 3.8+ Support

Compatible with Python 3.8+ with full async/await support and type hints.

Async/Await Native

Built from the ground up with asyncio for high-performance concurrent operations.

Type Safety

Comprehensive type hints for better IDE support and runtime safety.

High Performance

Optimized for low latency with efficient Socket.IO connections and smart routing.

Cost Effective

No per-message pricing, industry-standard 32KB message limits, transparent pricing.

Cross Platform

Works on Windows, macOS, and Linux with consistent behavior across platforms.

Installation

bash

pip install oddsockets

bash

poetry add oddsockets

bash

conda install -c conda-forge oddsockets

Quick Start

Basic Usage

python

import asyncio
from oddsockets import OddSockets

async def message_handler(data):
    print(f"Received: {data}")

async def main():
    # Create client
    client = OddSockets({
        'api_key': 'ak_live_1234567890abcdef'
    })
    
    # Get channel
    channel = client.channel('my-channel')
    
    # Subscribe to messages
    await channel.subscribe(message_handler)
    
    # Publish a message
    await channel.publish('Hello, World!')
    
    # Keep alive
    await asyncio.sleep(5)
    
    # Clean up
    await client.disconnect()

# Run the async function
asyncio.run(main())

Synchronous Wrapper

python

# For applications that need synchronous API
from oddsockets.sync import OddSockets

def message_handler(data):
    print(f"Received: {data}")

# Create client
client = OddSockets({
    'api_key': 'ak_live_1234567890abcdef'
})

# Get channel
channel = client.channel('my-channel')

# Subscribe to messages
channel.subscribe(message_handler)

# Publish a message
channel.publish('Hello, World!')

# Clean up
client.disconnect()

Type Hints Usage

python

import asyncio
from typing import Dict, Any
from oddsockets import OddSockets, Channel

async def typed_message_handler(data: Dict[str, Any]) -> None:
    print(f"Received: {data}")

async def main() -> None:
    client: OddSockets = OddSockets({
        'api_key': 'ak_live_1234567890abcdef'
    })
    
    channel: Channel = client.channel('typed-channel')
    await channel.subscribe(typed_message_handler)

asyncio.run(main())

Configuration

Client Options

python

client = OddSockets({
    'api_key': 'your-api-key',           # Required: Your OddSockets API key
    'user_id': 'user-id',                # Optional: User identifier
    'auto_connect': True,                # Optional: Auto-connect on creation
    'options': {
        'reconnect_attempts': 5,         # Optional: Max reconnection attempts
        'heartbeat_interval': 30.0       # Optional: Heartbeat interval (seconds)
    }
})

Channel Options

python

await channel.subscribe(callback, {
    'enable_presence': True,             # Enable presence tracking
    'retain_history': True,              # Retain message history
    'max_history': 100                   # Maximum history messages
})

await channel.publish(message, {
    'ttl': 3600,                         # Time to live (seconds)
    'metadata': {'priority': 'high'},    # Additional metadata
})

Examples

Explore comprehensive examples demonstrating the OddSockets Python SDK in action:

View All Examples

Browse all available examples including async/await code, framework integrations, and interactive tutorials

Performance & Compatibility

OddSockets Python SDK delivers superior performance with broad compatibility:

<50ms

Latency

99.9%

Uptime

32KB

Max Message

1M+

Messages/sec

Python Support

Python 3.8+ (2019)
Python 3.9+ (2020)
Python 3.10+ (2021)
Python 3.11+ (2022)
Python 3.12+ (2023)

Platform Support

Windows 10+
macOS 10.15+
Linux (Ubuntu 18.04+)
Docker containers

Framework Integrations

The OddSockets Python SDK works seamlessly with all modern Python frameworks. Here are examples showing how to integrate with popular frameworks:

FastAPI

python

from fastapi import FastAPI, WebSocket
from oddsockets import OddSockets
import asyncio

app = FastAPI()

# Initialize OddSockets client
client = OddSockets({
    'api_key': 'ak_live_1234567890abcdef'
})

@app.on_event("startup")
async def startup_event():
    # Connect to OddSockets on startup
    await client.connect()

@app.on_event("shutdown")
async def shutdown_event():
    # Disconnect on shutdown
    await client.disconnect()

@app.websocket("/ws/{channel_name}")
async def websocket_endpoint(websocket: WebSocket, channel_name: str):
    await websocket.accept()
    
    # Get OddSockets channel
    channel = client.channel(channel_name)
    
    # Message handler
    async def on_message(data):
        await websocket.send_json(data)
    
    # Subscribe to channel
    await channel.subscribe(on_message)
    
    try:
        while True:
            # Receive message from WebSocket
            data = await websocket.receive_json()
            # Publish to OddSockets channel
            await channel.publish(data)
    except Exception:
        await channel.unsubscribe()
        await websocket.close()

Django Channels

python

from channels.generic.websocket import AsyncWebsocketConsumer
from oddsockets import OddSockets
import json

class ChatConsumer(AsyncWebsocketConsumer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = OddSockets({
            'api_key': 'ak_live_1234567890abcdef'
        })
        self.channel = None

    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        
        # Accept WebSocket connection
        await self.accept()
        
        # Get OddSockets channel
        self.channel = self.client.channel(f'chat_{self.room_name}')
        
        # Subscribe to messages
        await self.channel.subscribe(self.on_oddsockets_message)

    async def disconnect(self, close_code):
        # Unsubscribe from OddSockets
        if self.channel:
            await self.channel.unsubscribe()
        await self.client.disconnect()

    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']
        
        # Publish to OddSockets
        await self.channel.publish({
            'message': message,
            'user': self.scope['user'].username
        })

    async def on_oddsockets_message(self, data):
        # Send message to WebSocket
        await self.send(text_data=json.dumps({
            'message': data['message'],
            'user': data['user']
        }))

Flask-SocketIO

python

from flask import Flask
from flask_socketio import SocketIO, emit, join_room, leave_room
from oddsockets import OddSockets
import asyncio
import threading

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")

# Initialize OddSockets client
client = OddSockets({
    'api_key': 'ak_live_1234567890abcdef'
})

# Run async client in separate thread
def run_async_client():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(client.connect())
    loop.run_forever()

# Start async client thread
client_thread = threading.Thread(target=run_async_client, daemon=True)
client_thread.start()

@socketio.on('join')
def on_join(data):
    room = data['room']
    join_room(room)
    
    # Get OddSockets channel
    channel = client.channel(f'room_{room}')
    
    # Message handler
    def on_message(msg_data):
        socketio.emit('message', msg_data, room=room)
    
    # Subscribe to channel (run in async context)
    asyncio.run_coroutine_threadsafe(
        channel.subscribe(on_message), 
        client._loop
    )

@socketio.on('message')
def handle_message(data):
    room = data['room']
    message = data['message']
    
    # Get channel and publish
    channel = client.channel(f'room_{room}')
    asyncio.run_coroutine_threadsafe(
        channel.publish(message), 
        client._loop
    )

if __name__ == '__main__':
    socketio.run(app, debug=True)

Quart (Async Flask)

python

from quart import Quart, websocket
from oddsockets import OddSockets
import asyncio
import json

app = Quart(__name__)

# Initialize OddSockets client
client = OddSockets({
    'api_key': 'ak_live_1234567890abcdef'
})

@app.before_serving
async def startup():
    await client.connect()

@app.after_serving
async def shutdown():
    await client.disconnect()

@app.websocket('/ws/')
async def ws(channel_name):
    # Get OddSockets channel
    channel = client.channel(channel_name)
    
    # Message handler
    async def on_message(data):
        await websocket.send(json.dumps(data))
    
    # Subscribe to channel
    await channel.subscribe(on_message)
    
    try:
        while True:
            # Receive message from WebSocket
            message = await websocket.receive()
            data = json.loads(message)
            
            # Publish to OddSockets channel
            await channel.publish(data)
    except Exception:
        await channel.unsubscribe()

if __name__ == '__main__':
    app.run()

Tornado

python

import tornado.ioloop
import tornado.web
import tornado.websocket
from oddsockets import OddSockets
import json
import asyncio

class ChatWebSocketHandler(tornado.websocket.WebSocketHandler):
    def initialize(self):
        self.client = OddSockets({
            'api_key': 'ak_live_1234567890abcdef'
        })
        self.channel = None

    async def open(self, channel_name):
        print(f"WebSocket opened for channel: {channel_name}")
        
        # Connect to OddSockets
        await self.client.connect()
        
        # Get channel
        self.channel = self.client.channel(channel_name)
        
        # Subscribe to messages
        await self.channel.subscribe(self.on_oddsockets_message)

    async def on_message(self, message):
        try:
            data = json.loads(message)
            # Publish to OddSockets
            await self.channel.publish(data)
        except json.JSONDecodeError:
            await self.write_message({
                'error': 'Invalid JSON message'
            })

    async def on_oddsockets_message(self, data):
        # Send message to WebSocket client
        await self.write_message(data)

    async def on_close(self):
        print("WebSocket closed")
        if self.channel:
            await self.channel.unsubscribe()
        await self.client.disconnect()

    def check_origin(self, origin):
        return True  # Allow all origins for demo

def make_app():
    return tornado.web.Application([
        (r"/ws/([^/]+)", ChatWebSocketHandler),
    ])

if __name__ == "__main__":
    app = make_app()
    app.listen(8888)
    print("Server started on http://localhost:8888")
    tornado.ioloop.IOLoop.current().start()