Socket Programming
socket is Python’s standard library module for low-level network communication. All higher-level networking libraries (http, asyncio, aiohttp, etc.) are built on top of socket.
Socket Basic Functions
| Function | Description |
|---|---|
socket.socket(family, type) | Create a socket. AF_INET=IPv4, SOCK_STREAM=TCP, SOCK_DGRAM=UDP |
s.bind((host, port)) | Bind an address and port (server side) |
s.listen(backlog) | Start listening; backlog is the length of the pending connection queue |
s.accept() | Block and wait for a client connection; returns (conn, addr) |
s.connect((host, port)) | Connect to the server (client side) |
s.send(data) | Send byte data |
s.recv(bufsize) | Receive up to bufsize bytes |
s.close() | Close the socket |
s.setsockopt(level, option, value) | Set socket options |
s.settimeout(seconds) | Set a timeout |
TCP Communication Example
TCP is a reliable, connection-oriented protocol. A three-way handshake must establish a connection before communication begins.
TCP Server
import socket
def run_server(host: str = "127.0.0.1", port: int = 9000) -> None:
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# SO_REUSEADDR: allow reuse of the port to avoid "address already in use" errors from TIME_WAIT state
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((host, port))
server.listen(5)
print(f"Server listening on {host}:{port}")
while True:
conn, addr = server.accept() # Block and wait for a new connection
print(f"Client connected: {addr}")
try:
while True:
data = conn.recv(1024)
if not data:
break # Client disconnected
message = data.decode("utf-8")
print(f"Received: {message}")
reply = f"Server received: {message}"
conn.sendall(reply.encode("utf-8"))
finally:
conn.close()
print(f"Client {addr} disconnected")
if __name__ == "__main__":
run_server()TCP Client
import socket
def run_client(host: str = "127.0.0.1", port: int = 9000) -> None:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
client.connect((host, port))
print(f"Connected to {host}:{port}")
messages = ["Hello!", "Python socket", "Goodbye"]
for msg in messages:
client.sendall(msg.encode("utf-8"))
data = client.recv(1024)
print(f"Server reply: {data.decode('utf-8')}")
if __name__ == "__main__":
run_client()Handling Multiple Clients (Multi-threading)
A single-process server can only handle one client at a time. Using multiple threads allows concurrent handling of multiple connections:
import socket
import threading
def handle_client(conn: socket.socket, addr: tuple) -> None:
print(f"New connection: {addr}")
try:
while True:
data = conn.recv(1024)
if not data:
break
conn.sendall(data) # Echo server: send data back as-is
finally:
conn.close()
def run_echo_server(host: str = "0.0.0.0", port: int = 9001) -> None:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((host, port))
server.listen(10)
print(f"Echo server started on {host}:{port}")
while True:
conn, addr = server.accept()
t = threading.Thread(target=handle_client, args=(conn, addr), daemon=True)
t.start()
if __name__ == "__main__":
run_echo_server()UDP Communication Example
UDP is connectionless and does not guarantee delivery, but it is fast and suitable for real-time scenarios (video streaming, gaming, etc.).
UDP Server
import socket
def run_udp_server(host: str = "127.0.0.1", port: int = 9002) -> None:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server:
server.bind((host, port))
print(f"UDP server listening on {host}:{port}")
while True:
data, addr = server.recvfrom(1024) # Returns both data and the sender's address
print(f"From {addr}: {data.decode()}")
reply = f"Received: {data.decode()}"
server.sendto(reply.encode(), addr)
if __name__ == "__main__":
run_udp_server()UDP Client
import socket
def run_udp_client(host: str = "127.0.0.1", port: int = 9002) -> None:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
client.settimeout(5) # 5-second timeout
for msg in ["Hello", "World"]:
client.sendto(msg.encode(), (host, port))
try:
reply, server_addr = client.recvfrom(1024)
print(f"Server reply: {reply.decode()}")
except socket.timeout:
print("Timed out waiting for reply")
if __name__ == "__main__":
run_udp_client()Timeouts and Non-blocking
import socket
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("127.0.0.1", 9003))
server.listen(5)
# Set a timeout for accept
server.settimeout(10)
try:
conn, addr = server.accept()
print(f"Connection: {addr}")
except socket.timeout:
print("No client connected within 10 seconds, exiting")
finally:
server.close()Getting Local Host Information
import socket
hostname = socket.gethostname()
local_ip = socket.gethostbyname(hostname)
print(f"Hostname: {hostname}")
print(f"Local IP: {local_ip}")
# Look up a remote host's IP
remote_ip = socket.gethostbyname("www.python.org")
print(f"python.org IP: {remote_ip}")Last updated on