How To Detect If A Date Is A Public Holiday In Python 3

Intermediate

You build a web app that needs live updates — a chat interface, a dashboard that refreshes when data changes, a multiplayer game, or a notification feed. You try the obvious approach: poll the server every two seconds with setInterval(fetch(...)). It works, but now your server is drowning in requests for data that has not changed, your latency is measured in seconds instead of milliseconds, and your browser tab is burning CPU even when nothing is happening. Polling is the traffic jam of web architecture.

Python’s python-socketio library solves this with a proper two-way connection. The browser and server stay connected and push events to each other the moment something changes — no polling, no wasted requests. Socket.IO builds on WebSockets but adds automatic fallbacks, reconnection logic, namespaces, and rooms out of the box. Install it with pip install python-socketio aiohttp and you have everything you need to build a real-time server in Python.

This article covers the complete socket.io toolkit for Python: setting up an async server with aiohttp, emitting and listening to events from both server and client, using rooms to group connections, broadcasting to multiple clients, and a real-life example that builds a live notification server. By the end, you will be able to push events from your Python backend to any connected browser the moment they occur.

Python socketio: Quick Example

The fastest way to understand socket.io is to run a server that echoes a message back to whoever sent it. This requires two files: a Python server and an HTML page that connects to it.

# server_quick.py
import asyncio
import aiohttp
from aiohttp import web
import socketio

# Create the Socket.IO server
sio = socketio.AsyncServer(cors_allowed_origins='*')
app = web.Application()
sio.attach(app)

@sio.event
async def connect(sid, environ):
    print(f'Client connected: {sid}')

@sio.event
async def disconnect(sid):
    print(f'Client disconnected: {sid}')

@sio.event
async def message(sid, data):
    print(f'Received from {sid}: {data}')
    # Echo it back to the sender only
    await sio.emit('reply', {'text': f'Server got: {data}'}, to=sid)

if __name__ == '__main__':
    web.run_app(app, port=8080)

Output (server terminal):

======== Running on http://0.0.0.0:8080 ========
Client connected: abc123def456
Received from abc123def456: Hello server
Client disconnected: abc123def456

The @sio.event decorator registers a function as a handler for a named event. The connect and disconnect events are built-in — they fire automatically whenever a client joins or leaves. The message event is custom — you define the name and the handler. When a browser sends a message event, your Python function receives the session ID (sid) and the data payload, and can emit a response back instantly using sio.emit().

The real power comes when you start emitting to multiple clients simultaneously, organizing connections into rooms, and broadcasting from server to browser without waiting for a client request. The sections below cover all of that in detail.

What Is socket.io and Why Use It?

Socket.IO is a protocol and library that provides reliable, bidirectional, event-based communication between a browser and a server. It was originally a JavaScript library, and python-socketio is the official Python implementation — it is fully compatible with the JavaScript socket.io client library used in browsers. When a browser connects to your Python socket.io server, both sides can emit named events with JSON payloads at any time, in either direction, without either side having to “ask” first.

The key distinction from raw WebSockets is that Socket.IO adds a protocol layer on top. It handles connection upgrade (from HTTP to WebSocket), automatic reconnection with exponential backoff, event acknowledgements, namespaces (logical separation within one connection), and rooms (groups of connections that can be addressed together). You get these features for free without building them yourself.

Feature	HTTP Polling	Raw WebSocket	Socket.IO
Direction	Client initiates only	Both directions	Both directions
Latency	Poll interval (1-30s)	Near-instant	Near-instant
Auto-reconnect	Manual	Manual	Built in
Rooms / groups	Not available	Manual	Built in
Namespaces	Not available	Not available	Built in
Fallback (no WS)	N/A	Not available	Falls back to polling
Browser client	fetch / XMLHttpRequest	WebSocket API	socket.io-client JS lib

Install the library with pip. The aiohttp package provides the async HTTP server that socket.io attaches to. You can also use flask-socketio if you prefer Flask’s synchronous model, but the async approach scales better for concurrent connections.

# Install in your terminal
# pip install python-socketio aiohttp

# Verify installation
import socketio
print(socketio.__version__)  # e.g., 5.11.0

The socketio.AsyncServer class is the async variant — use it with aiohttp or any ASGI framework. The socketio.Server class is the synchronous variant for use with Flask or WSGI apps. This article uses the async version throughout, since it is the recommended approach for new projects.

Setting Up an Async Socket.IO Server

A production-ready socket.io server needs three things: the AsyncServer instance, an aiohttp web application to serve HTTP alongside the socket.io traffic, and a static HTML file for the browser client. Here is a complete server with all three, including a health check endpoint so you can verify the server is running from a browser.

# server_full.py
import aiohttp
from aiohttp import web
import socketio

# cors_allowed_origins controls which browser origins can connect.
# Use '*' in development, list specific domains in production.
sio = socketio.AsyncServer(
    cors_allowed_origins='*',
    logger=True,
    engineio_logger=False  # set True to see low-level ping/pong traffic
)

app = web.Application()
sio.attach(app)

# Standard aiohttp route alongside socket.io
async def index(request):
    with open('client.html') as f:
        return web.Response(text=f.read(), content_type='text/html')

app.router.add_get('/', index)
app.router.add_static('/static', path='./static', name='static')

@sio.event
async def connect(sid, environ, auth):
    # auth is the data passed in the socket.io connect options from the client
    print(f'[CONNECT] sid={sid}  ip={environ.get("REMOTE_ADDR")}  auth={auth}')

@sio.event
async def disconnect(sid):
    print(f'[DISCONNECT] sid={sid}')

@sio.event
async def ping_server(sid, data):
    """Client sends ping_server; server responds with pong_client."""
    print(f'[PING] from {sid}: {data}')
    await sio.emit('pong_client', {'received': data, 'from': sid}, to=sid)

if __name__ == '__main__':
    web.run_app(app, host='0.0.0.0', port=8080)

Output (server startup):

======== Running on http://0.0.0.0:8080 ========
(Press CTRL+C to quit)

The cors_allowed_origins='*' setting is important during development — without it, the browser will block the WebSocket upgrade with a CORS error. In production, replace '*' with a list of your actual domains: ['https://myapp.com', 'https://www.myapp.com']. The logger=True flag sends socket.io connection events to the standard Python logger, which is useful for debugging connection issues.

The auth parameter in the connect handler deserves attention. The browser client can pass authentication data (a token, a user ID) at connection time — your server receives it in auth and can reject the connection by returning False. This is the right place to validate tokens before accepting a client.

The Browser Client

The browser side uses the official socket.io-client JavaScript library, loaded from a CDN. It mirrors the Python API exactly — socket.on() listens for events, socket.emit() sends them.

<!-- client.html -->
<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <title>Socket.IO Demo</title>
</head>
<body>
  <h2>Socket.IO Demo</h2>
  <div id="status">Connecting...</div>
  <button onclick="sendPing()">Send Ping</button>
  <ul id="log"></ul>

  <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.7.2/socket.io.min.js"></script>
  <script>
    const socket = io('http://localhost:8080', {
      auth: { token: 'demo-user-123' }
    });

    socket.on('connect', () => {
      document.getElementById('status').textContent = 'Connected: ' + socket.id;
    });

    socket.on('disconnect', () => {
      document.getElementById('status').textContent = 'Disconnected';
    });

    // Listen for pong_client events from the server
    socket.on('pong_client', (data) => {
      const li = document.createElement('li');
      li.textContent = 'Server replied: ' + JSON.stringify(data);
      document.getElementById('log').appendChild(li);
    });

    function sendPing() {
      socket.emit('ping_server', { time: Date.now(), msg: 'hello' });
    }
  </script>
</body>
</html>

The pattern mirrors the Python server exactly. socket.on('event_name', handler) registers a listener for any event the server emits. socket.emit('event_name', data) sends an event to the server. The connection is established automatically when the page loads — you do not call any “connect” function explicitly. The auth option passes data that arrives in the Python connect handler’s auth parameter, where you can validate it.

Rooms and Broadcasting

Rooms are how socket.io handles group communication. A room is simply a named bucket that connections can join and leave. When you emit to a room, every connection in that room receives the event — without you having to track which clients are in which group. This is the foundation for features like chat channels, topic subscriptions, and per-user notification queues.

# server_rooms.py
import aiohttp
from aiohttp import web
import socketio

sio = socketio.AsyncServer(cors_allowed_origins='*')
app = web.Application()
sio.attach(app)

@sio.event
async def connect(sid, environ):
    print(f'Connected: {sid}')

@sio.event
async def disconnect(sid):
    print(f'Disconnected: {sid}')

@sio.event
async def join_room(sid, data):
    """Client sends join_room with {'room': 'sports'} to subscribe."""
    room = data.get('room')
    if not room:
        return
    await sio.enter_room(sid, room)
    print(f'{sid} joined room: {room}')
    # Notify the joining client only
    await sio.emit('room_joined', {'room': room, 'sid': sid}, to=sid)

@sio.event
async def leave_room(sid, data):
    """Client sends leave_room with {'room': 'sports'} to unsubscribe."""
    room = data.get('room')
    if not room:
        return
    await sio.leave_room(sid, room)
    print(f'{sid} left room: {room}')

@sio.event
async def send_to_room(sid, data):
    """Broadcast a message to everyone in a room (excluding the sender)."""
    room = data.get('room')
    message = data.get('message', '')
    if not room or not message:
        return
    # skip_sid=sid means the sender does not receive their own broadcast
    await sio.emit(
        'room_message',
        {'from': sid, 'room': room, 'message': message},
        room=room,
        skip_sid=sid
    )

async def server_broadcast(app):
    """Background task: push a server-initiated event every 10 seconds."""
    import asyncio
    count = 0
    while True:
        await asyncio.sleep(10)
        count += 1
        await sio.emit('server_tick', {'count': count, 'msg': 'Heartbeat from server'})
        print(f'Broadcast tick #{count} to all clients')

app.on_startup.append(lambda app: asyncio.ensure_future(server_broadcast(app)))

if __name__ == '__main__':
    import asyncio
    web.run_app(app, port=8080)

Server output when two clients join “sports” and one sends a message:

Connected: sid_client_A
Connected: sid_client_B
sid_client_A joined room: sports
sid_client_B joined room: sports
Broadcast tick #1 to all clients
sid_client_A left room: sports

Four functions do all the room work. sio.enter_room(sid, room) adds a connection to a room. sio.leave_room(sid, room) removes it. sio.emit('event', data, room='roomname') sends to everyone in the room. The skip_sid parameter is the socket.io idiom for “broadcast to the room but not the sender” — exactly what you want for chat-style features where the sender already knows what they typed. Notice also the background server_broadcast coroutine: it emits a server_tick event to all connected clients every 10 seconds without any client request. This is how server-initiated push notifications work.

Namespaces: Logical Separation Within One Connection

Namespaces let you split your socket.io traffic into separate logical channels over a single physical WebSocket connection. Think of namespaces like URL paths for events: a client connected to /admin only receives events emitted to that namespace. This is cleaner than inventing your own event-name prefixes and is the right tool when you have genuinely separate concerns — such as a public feed and an admin control panel — sharing the same server.

# server_namespaces.py
import aiohttp
from aiohttp import web
import socketio

sio = socketio.AsyncServer(cors_allowed_origins='*')
app = web.Application()
sio.attach(app)

# Default namespace '/' -- open to everyone
@sio.event
async def connect(sid, environ):
    print(f'[/] Connected: {sid}')

@sio.event
async def public_message(sid, data):
    print(f'[/] Public message from {sid}: {data}')
    await sio.emit('public_reply', {'msg': 'Hello from public namespace'}, to=sid)

# Admin namespace '/admin' -- you would validate credentials in connect
@sio.event(namespace='/admin')
async def connect(sid, environ):
    print(f'[/admin] Admin connected: {sid}')

@sio.event(namespace='/admin')
async def admin_action(sid, data):
    action = data.get('action', 'unknown')
    print(f'[/admin] Action from {sid}: {action}')
    # Broadcast result to ALL admin clients
    await sio.emit('admin_result', {'action': action, 'status': 'done'}, namespace='/admin')

if __name__ == '__main__':
    web.run_app(app, port=8080)

Browser client connecting to the admin namespace:

// In the browser JavaScript
const publicSocket = io('http://localhost:8080/');
const adminSocket  = io('http://localhost:8080/admin');

adminSocket.on('connect', () => console.log('Admin connected:', adminSocket.id));
adminSocket.on('admin_result', (data) => console.log('Admin result:', data));
adminSocket.emit('admin_action', { action: 'purge_cache' });

The namespace='/admin' argument on @sio.event scopes that handler exclusively to the admin channel. Events emitted on / (the default namespace) are invisible to clients connected on /admin, and vice versa. In a real app, you would validate a token in the /admin connect handler and return False to reject unauthorized connections before they ever reach the event handlers.

Using the Python Client

Socket.IO is not just for browser clients. The python-socketio library also ships an async client, which lets you connect one Python process to another as equals — useful for microservices that need real-time coordination, or for writing integration tests against your socket.io server.

# client_python.py
import asyncio
import socketio

# Async Python client
sio_client = socketio.AsyncClient()

@sio_client.event
async def connect():
    print('Python client connected, sid:', sio_client.sid)

@sio_client.event
async def disconnect():
    print('Python client disconnected')

@sio_client.event
async def pong_client(data):
    print('Received from server:', data)

async def main():
    await sio_client.connect('http://localhost:8080', auth={'token': 'python-client'})

    # Send a ping event and wait a moment for the reply
    await sio_client.emit('ping_server', {'msg': 'hello from Python client'})
    await asyncio.sleep(1)

    # Join a room
    await sio_client.emit('join_room', {'room': 'sports'})
    await asyncio.sleep(1)

    await sio_client.disconnect()

if __name__ == '__main__':
    asyncio.run(main())

Output:

Python client connected, sid: xyz789abc000
Received from server: {'received': {'msg': 'hello from Python client'}, 'from': 'xyz789abc000'}

The async Python client mirrors the server API. Decorate handlers with @sio_client.event and connect with await sio_client.connect(url). You can use it to write automated tests against your socket.io server, to connect two backend services for real-time coordination, or to build a bot that responds to server-emitted events. The same reconnection and room logic applies — the client and server are peers speaking the same protocol.

Real-Life Example: Live Notification Server

This project builds a notification hub: a Python server that accepts incoming “alert” events from any Python process (a monitoring script, a background worker, a webhook handler) and broadcasts them instantly to all connected browser clients. This is the core pattern behind live dashboards, build status monitors, and ops alert feeds.

# notification_server.py
import asyncio
import aiohttp
from aiohttp import web
import socketio
import datetime

sio = socketio.AsyncServer(cors_allowed_origins='*', logger=False)
app = web.Application()
sio.attach(app)

# Track connected clients and their subscribed categories
connected_clients = {}  # sid -> {'categories': set()}

@sio.event
async def connect(sid, environ, auth):
    connected_clients[sid] = {'categories': set(), 'connected_at': datetime.datetime.utcnow().isoformat()}
    total = len(connected_clients)
    print(f'[+] {sid} connected. Total: {total}')
    await sio.emit('welcome', {'sid': sid, 'total_clients': total}, to=sid)

@sio.event
async def disconnect(sid):
    connected_clients.pop(sid, None)
    print(f'[-] {sid} disconnected. Total: {len(connected_clients)}')

@sio.event
async def subscribe(sid, data):
    """Client subscribes to one or more notification categories."""
    categories = data.get('categories', [])
    if not isinstance(categories, list):
        categories = [categories]
    for cat in categories:
        cat = str(cat).lower().strip()
        if cat:
            await sio.enter_room(sid, f'cat_{cat}')
            connected_clients[sid]['categories'].add(cat)
    await sio.emit('subscribed', {'categories': list(connected_clients[sid]['categories'])}, to=sid)
    print(f'[SUB] {sid} subscribed to: {categories}')

@sio.event
async def unsubscribe(sid, data):
    """Client unsubscribes from a category."""
    category = data.get('category', '').lower().strip()
    if category and category in connected_clients.get(sid, {}).get('categories', set()):
        await sio.leave_room(sid, f'cat_{category}')
        connected_clients[sid]['categories'].discard(category)

@sio.event
async def push_alert(sid, data):
    """Any client (e.g. a monitoring script) can push an alert to a category room."""
    category = str(data.get('category', 'general')).lower().strip()
    message = data.get('message', '')
    severity = data.get('severity', 'info')  # info | warning | error
    if not message:
        return

    alert_payload = {
        'category': category,
        'message': message,
        'severity': severity,
        'timestamp': datetime.datetime.utcnow().isoformat(),
        'from_sid': sid,
    }
    # Broadcast to everyone subscribed to this category (including sender)
    await sio.emit('alert', alert_payload, room=f'cat_{category}')
    print(f'[ALERT] {severity.upper()} in {category}: {message}')

async def status_handler(request):
    """HTTP endpoint: returns JSON status of connected clients."""
    clients_info = [
        {'sid': sid, 'categories': list(info['categories'])}
        for sid, info in connected_clients.items()
    ]
    return web.json_response({'clients': clients_info, 'total': len(connected_clients)})

app.router.add_get('/status', status_handler)

if __name__ == '__main__':
    web.run_app(app, host='0.0.0.0', port=8080)

A monitoring script that sends alerts to the server:

# alert_sender.py
import asyncio
import socketio

async def send_alert(category, message, severity='info'):
    client = socketio.AsyncClient()
    await client.connect('http://localhost:8080')
    await client.emit('push_alert', {
        'category': category,
        'message': message,
        'severity': severity,
    })
    await asyncio.sleep(0.5)  # wait for the emit to flush
    await client.disconnect()

async def main():
    await send_alert('deploy', 'v2.3.1 deployed to production', severity='info')
    await send_alert('database', 'Connection pool at 95% capacity', severity='warning')
    await send_alert('payments', 'Stripe webhook failed 3 times', severity='error')

if __name__ == '__main__':
    asyncio.run(main())

Server output:

[+] abc123 connected. Total: 1   # browser tab
[+] xyz456 connected. Total: 2   # alert_sender.py
[ALERT] INFO in deploy: v2.3.1 deployed to production
[ALERT] WARNING in database: Connection pool at 95% capacity
[ALERT] ERROR in payments: Stripe webhook failed 3 times
[-] xyz456 disconnected. Total: 1

The server separates two roles cleanly: browser clients subscribe to categories and receive alert events; monitoring scripts connect briefly, push alerts via push_alert, and disconnect. The /status HTTP endpoint gives you a JSON view of who is connected and what they are subscribed to — useful for an ops dashboard sidebar showing active watchers. To extend this, add a Redis adapter (pip install python-socketio[asyncio_client] aioredis) so the server can scale across multiple worker processes while still broadcasting to all connected clients.

Frequently Asked Questions

What is the difference between WebSocket and socket.io?

WebSocket is a browser protocol that establishes a persistent two-way connection. Socket.IO is a library that uses WebSocket as its transport but adds a structured protocol layer on top: named events, automatic reconnection, rooms, namespaces, and fallback to HTTP long-polling when WebSocket is blocked. If you use raw WebSocket, you must implement all of that yourself. Socket.IO gives you those features out of the box, which is why most real-time apps reach for it rather than raw WebSocket. The trade-off is that the socket.io client and server must speak the same protocol version — you cannot connect a browser’s native WebSocket object to a socket.io server without the socket.io client library.

How does socket.io scale across multiple server processes?

By default, python-socketio keeps room and connection state in memory, which means a client connected to worker A cannot receive an event emitted by worker B. To scale, you need a message queue adapter. The most common option is Redis: pip install python-socketio[asyncio_client] aioredis, then configure the server with socketio.AsyncServer(client_manager=socketio.AsyncAioPikaManager('redis://localhost')). With a Redis adapter, all workers share room state and broadcasts fan out across every process. This is the standard production pattern for running socket.io behind a load balancer.

How do I detect when a client disconnects unexpectedly?

The disconnect event fires for all disconnections — graceful logouts, network drops, browser tab closes, and timeouts. You do not need to distinguish between them to clean up state. Socket.IO uses a heartbeat mechanism (configurable via ping_interval and ping_timeout on AsyncServer) to detect stale connections: if a client does not respond to a ping within ping_timeout seconds, the server treats it as disconnected and fires the disconnect event. The default is a 5-second ping interval with a 20-second timeout. For gaming or trading apps that need faster detection, lower both values — but watch out for false positives on slow mobile networks.

How do I authenticate socket.io connections?

The standard pattern is to pass a token in the auth option of the client’s io()` call: `io('http://localhost:8080', { auth: { token: 'jwt-here' } }). The Python server receives this in the connect handler’s auth parameter. Validate the token there and return False to reject the connection. Connections rejected with False receive a connection error on the client side. Never trust event handlers for authentication — if a connection gets through connect, treat it as authenticated for the session. Token validation in every event handler is too slow and too fragile.

How do event acknowledgements work?

Socket.IO supports acknowledgements: the sender can pass a callback that the receiver calls after processing the event. In the Python async API, you request an acknowledgement using await sio.call('event_name', data, to=sid) (server side) or const result = await socket.emitWithAck('event_name', data) (browser side). The call blocks until the other side responds, giving you a request-response pattern over the event channel. This is useful for confirming that a command was received and acted on — for example, confirming that a room-join completed before the client starts sending messages into that room. Acknowledgements time out after a configurable period if the other side does not respond.

Conclusion

Python’s python-socketio library gives you a complete real-time event layer with almost no boilerplate. The core concepts covered in this article — the AsyncServer setup with aiohttp, the @sio.event decorator for named events, sio.enter_room() and sio.emit(room=...) for group broadcasts, namespaces for logical separation, and the async Python client for server-to-server communication — cover the full surface area of most real-time web applications.

The notification server project shows the most common production pattern: a Python backend that accepts push events from internal services and relays them to subscribed browser clients in real time. Extend it by adding Redis for multi-process scaling, JWT validation in the connect handler, and a React or Vue frontend that renders incoming alert payloads as styled cards. You have everything you need to build live dashboards, multiplayer features, or any application where waiting for a poll is a second too long.

For the full API reference — including acknowledgements, binary data transfer, and the ASGI adapter for use with FastAPI or Starlette — see the official python-socketio documentation and the Socket.IO protocol docs.

Related Articles

• How To Build a WebSocket Server with Python
• How To Use Python Celery with Redis for Task Queues
• How To Use Python asyncio for Concurrent Tasks