Python real-time with WebSockets

https://miro.medium.com/v2/resize:fit:1400/1*ori9BopVCAUTXhX_JBsV7Q.png

Have you ever built an app that needs real-time features — like a chat app, a live scoreboard, or a notification system? If yes, then WebSockets might be the best fit for your use case. In this article, we’ll explore what WebSockets are, how they work in Python, and when to use them.

What is WebSocket?

WebSocket is a protocol that enables two-way communication between the server and the client over a single, long-lived connection. Unlike HTTP which is request-response based, WebSocket allows servers to push data to clients without being asked.

Use Cases of WebSocket

• Real-time chat applications
• Online multiplayer games
• Live sports updates
• Financial tickers
• Notification systems

Python WebSocket Example

We’ll use the websockets library to create a simple multi-client chat server.

1. Install the library

pip install websockets

Get Grassroot Engineer’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

2. Create the Chat Server (with Server-side Input Broadcast)

import asyncio
import websockets

connected_clients = set()

async def handler(websocket):
    connected_clients.add(websocket)
    print("✅ A client connected.")
    try:
        async for message in websocket:
            # Send back to sender
            await websocket.send(f"You said: {message}")
            print(f"📩 Received from client: {message}")
            # Broadcast to others
            await asyncio.gather(*[
                client.send(f"Someone says: {message}")
                for client in connected_clients if client != websocket
            ])
    finally:
        connected_clients.remove(websocket)
        print("❌ A client disconnected.")

async def server_input():
    while True:
        try:
            # Use asyncio.to_thread to make input non-blocking
            msg = await asyncio.to_thread(input, "[Server] Type message to broadcast: ")
            if connected_clients:  # Check if there are any connected clients
                await asyncio.gather(*[
                    client.send(f"🟢 Server: {msg}")
                    for client in connected_clients
                ])
                print(f"📤 Server broadcast: {msg}")
            else:
                print("❗ No clients connected to receive message")
        except Exception as e:
            print(f"Error in server input: {e}")

async def main():
    server = await websockets.serve(handler, "localhost", 8765)
    print("✅ Server is running on ws://localhost:8765")
    print("Waiting for client connections...")
    
    # Run both server input handling and keep the server alive
    await asyncio.gather(
        server_input(),
        asyncio.Future()  # This future never completes, keeping the server running
    )

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

3. Create the Client

import asyncio
import websockets
import sys

async def chat():
    uri = "ws://localhost:8765"
    try:
        print(f"Connecting to server at {uri}...")
        async with websockets.connect(uri) as websocket:
            print("✅ Connected to server!")
            print("Start chatting (type messages and press Enter)")
            
            async def receive():
                try:
                    async for message in websocket:
                        print(f"\n📩 {message}")
                        print("You: ", end="", flush=True)  # Prompt again after receiving
                except websockets.exceptions.ConnectionClosed:
                    print("\n❌ Connection to server closed")
                    return
                except Exception as e:
                    print(f"\n❌ Error receiving message: {e}")
                    return

            async def send():
                try:
                    while True:
                        # Use asyncio.to_thread to make input non-blocking
                        msg = await asyncio.to_thread(input, "You: ")
                        if msg.lower() == "exit":
                            print("Exiting chat...")
                            sys.exit(0)
                        await websocket.send(msg)
                except websockets.exceptions.ConnectionClosed:
                    print("\n❌ Connection to server closed")
                    return
                except Exception as e:
                    print(f"\n❌ Error sending message: {e}")
                    return

            await asyncio.gather(receive(), send())
    except ConnectionRefusedError:
        print("❌ Could not connect to server. Is the server running?")
    except Exception as e:
        print(f"❌ Error: {e}")

if __name__ == "__main__":
    try:
        asyncio.run(chat())
    except KeyboardInterrupt:
        print("\nExiting chat client...")
        sys.exit(0)

4. Running by

# Run server first
python server.py

# Run client later
python client.py

5. See my code snippet in Github > Github

6. Explanation in Thai.

Explain codes

Running both server and client

Conclusion

WebSockets are perfect for real-time applications where instant communication is crucial. With Python’s websockets library, building a real-time server is both easy and effective. Try it out and see your applications come to life with real-time power!

Ref:
https://geekpython.in/build-websocket-server-and-client-using-python

If you think it’s useful for you, just clap your hands 👏 to be encouraged me.
GRASSROOT ENGINEER 😘