Socket Programming - Python Notebook

Socket programming is a way to connect two nodes on a network to communicate with each other. One socket (node) listens on a particular port at an IP, while the other socket reaches out to the other to form a connection. The server forms the listener socket while the client reaches out to the server.

Simple Calculator¶

Let’s build a simple calculator using socket programming. The server will perform basic arithmetic operations like addition, subtraction, multiplication, and division based on the client’s request.

First let’s import socket and threading libraries. The socket library provides low-level networking interface, while the threading library allows us to handle multiple clients simultaneously.

import socket
import threading

Now a basic function to calculate the result based on the operation requested by the client.

def calculate(expression):
    """Evaluate a simple arithmetic expression"""
    try:
        if not all(c in "0123456789+-*/. " for c in expression):
            return "Error: Invalid characters in expression."

        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

Basic server (single client)¶

First let’s build a basic server that can handle a single client. The server will listen for incoming connections, receive the operation and numbers from the client, perform the calculation, and send back the result.

def start_server(host="127.0.0.1", port=9999):
    """Start the calculator server"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((host, port))
        s.listen()
        print(f"Calculator server listening on {host}:{port}")
        conn, addr = s.accept()
        with conn:
            print("Connected by", addr)
            while True:
                data = conn.recv(1024)
                if not data:
                    break
                expression = data.decode("utf-8")
                result = calculate(expression)
                conn.sendall(result.encode("utf-8"))

Basic Server (Multiple Clients)¶

We can enhance the server to handle multiple clients simultaneously using threading. Each client connection will be handled in a separate thread.

def start_threaded_server(host="127.0.0.1", port=9999):
    """Start the calculator server with threading support"""
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind((host, port))
    server_socket.listen()
    print(f"Threaded calculator server listening on {host}:{port}")

    def handle_client(conn, addr):
        print("Connected by", addr)
        with conn:
            while True:
                data = conn.recv(1024)
                if not data:
                    break
                expression = data.decode("utf-8")
                result = calculate(expression)
                conn.sendall(result.encode("utf-8"))
        print("Disconnected from", addr)

    try:
        while True:
            conn, addr = server_socket.accept()
            client_thread = threading.Thread(target=handle_client, args=(conn, addr))
            client_thread.daemon = True
            client_thread.start()
    except KeyboardInterrupt:
        print("\nShutting down server.")
    finally:
        server_socket.close()

And finally call the server function to start the server.

start_threaded_server()

Client Code¶

And finally the client code to connect to the server and send requests.

def client_program(host="127.0.0.1", port=9999):
    """Start a simple calculator client"""

    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        try:
            s.connect((host, port))
            print(f"✅ Connected to calculator server at {host}:{port}")

            while True:
                expression = input("Enter expression (or 'exit' to quit): ")
                if expression.lower() == "exit":
                    print("👋 Exiting calculator client.")
                    break

                s.sendall(expression.encode("utf-8"))
                data = s.recv(1024)
                print(f"Result: {data.decode('utf-8')}")

        except ConnectionRefusedError:
            print(
                f"❌ Could not connect to server at {host}:{port}. Is the server running?"
            )
        except Exception as e:
            print(f"❌ An error occurred: {e}")