Python Sockets and Selectors

The essentials

Quick reference

One focused task per row. Jump to the related section for complete, working examples.

Use	Syntax	Examples
Resolve stream candidates	`addresses = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)`	View examples
Open a TCP connection	`sock = socket.create_connection((host, port), timeout=5.0)`	View examples
Close a socket reliably	`with socket.socket() as sock: use(sock)`	View examples
Send the full buffer	`sock.sendall(payload)`	View examples
Detect stream EOF	`chunk = sock.recv(65536)`	View examples
Encode a network-order length	`header = struct.pack('!I', len(payload))`	View examples
Bind an ephemeral loopback port	`listener.bind(('127.0.0.1', 0))`	View examples
Start listening	`listener.listen(128)`	View examples
Accept one connection	`connection, address = listener.accept()`	View examples
Set an operation timeout	`sock.settimeout(2.0)`	View examples
Enable non-blocking mode	`sock.setblocking(False)`	View examples
Register read interest	`selector.register(sock, selectors.EVENT_READ, data=state)`	View examples
Wait for readiness	`events = selector.select(timeout=1.0)`	View examples
Change event interest	`selector.modify(sock, selectors.EVENT_READ \| selectors.EVENT_WRITE, state)`	View examples
Send a UDP datagram	`sent = sock.sendto(payload, destination)`	View examples
Receive a UDP datagram	`payload, peer = sock.recvfrom(65507)`	View examples
Half-close the write side	`sock.shutdown(socket.SHUT_WR)`	View examples
Create safe client TLS defaults	`context = ssl.create_default_context()`	View examples

Sockets expose byte streams and datagrams, not application messages. Correct programs define framing, handle partial progress, apply deadlines, close ownership boundaries, and treat every peer-supplied length or address as untrusted. The selectors module supplies a portable readiness layer, while TLS belongs in ssl rather than an application-designed encryption scheme.

Step by step

Detailed examples

Resolve addresses and connect with bounded waiting

getaddrinfo() may return IPv4 and IPv6 candidates; iterate results or let create_connection() try them. Always set a finite timeout appropriate to the operation, and use a context manager so failed parsing still closes the descriptor. Hostname resolution and connection can fail independently and should retain their original OSError context.

Preserve address-family information

import socket

candidates = [
    (socket.AF_INET6, ("::1", 443, 0, 0)),
    (socket.AF_INET, ("127.0.0.1", 443)),
]
for family, address in candidates:
    print(family.name, address[0])

Output

AF_INET6 ::1
AF_INET 127.0.0.1

Back to quick reference ↑

Frame messages on TCP byte streams

TCP preserves byte order but not send() or recv() call boundaries. Define a length prefix, delimiter, or self-delimiting format; bound declared sizes before allocating. sendall() retries until all bytes are accepted or an error occurs. recv() returning b'' means orderly peer shutdown, while short non-empty reads are ordinary.

Encode and decode one bounded frame

import struct

payload = b"hello"
wire = struct.pack("!I", len(payload)) + payload
size = struct.unpack("!I", wire[:4])[0]
if size > 1024:
    raise ValueError("frame too large")
print(wire[4:4 + size].decode("ascii"))

Output

hello

Back to quick reference ↑

Bind listeners without leaking accepted sockets

A TCP server binds a local address, listens, then accepts connected sockets. Bind to a specific interface unless remote exposure is intentional; an empty host or 0.0.0.0 exposes every IPv4 interface. SO_REUSEADDR semantics differ across platforms and do not authorize multiple active servers. Close each accepted connection independently from the listening socket.

Validate an intentional listener address

import ipaddress

def listener_address(host: str, port: int) -> tuple[str, int]:
    if not ipaddress.ip_address(host).is_loopback:
        raise ValueError("external exposure requires approval")
    if not 0 <= port <= 65535:
        raise ValueError("invalid port")
    return host, port

print(listener_address("127.0.0.1", 8080))

Output

('127.0.0.1', 8080)

Back to quick reference ↑

Distinguish deadlines from readiness

settimeout(seconds) makes operations raise TimeoutError after bounded waiting; setblocking(False) makes them report would-block immediately. A readiness notification is only a hint because another consumer or protocol state may intervene, so non-blocking operations must still handle BlockingIOError. Avoid mixing file-like wrappers with timeout mode unless buffering behavior is carefully controlled.

Classify non-blocking and timeout errors

import socket

would_block = BlockingIOError(11, "operation would block")
print(isinstance(would_block, OSError))
print(socket.timeout is TimeoutError)

Output

True
True

Back to quick reference ↑

Drive many sockets with DefaultSelector

DefaultSelector chooses the most capable portable backend available. Register each non-blocking socket with an event mask and application state, process the returned key/mask pairs, and modify interest when output buffers empty. Readiness loops need fairness, per-connection buffer limits, and explicit unregister-before-close cleanup.

Interpret selector event masks

import selectors

mask = selectors.EVENT_READ | selectors.EVENT_WRITE
state = {"input": bytearray(), "output": bytearray(b"reply")}
print(bool(mask & selectors.EVENT_READ))
print(bool(mask & selectors.EVENT_WRITE), len(state["output"]))

Output

True
True 5

Back to quick reference ↑

Preserve UDP datagram boundaries

UDP recvfrom() returns one datagram and its source address; an undersized buffer discards the remainder on common platforms. Datagrams may be lost, duplicated, reordered, or spoofed. Keep payloads below the path MTU, validate source and content, and build explicit retransmission and idempotency only when the application needs them.

Validate one complete datagram

def parse_datagram(payload: bytes, maximum: int = 1200) -> str:
    if len(payload) > maximum:
        raise ValueError("datagram too large")
    return payload.decode("utf-8", errors="strict")

packets = [b"one packet", b"second packet"]
print([parse_datagram(packet) for packet in packets])

Output

['one packet', 'second packet']

Back to quick reference ↑

Close protocols deliberately and add TLS

shutdown(SHUT_WR) signals that no more bytes will be sent while allowing remaining input to drain; close() releases the descriptor. For internet protocols, wrap sockets with an ssl.SSLContext configured for client or server use, verify hostnames, and never disable certificate validation as a workaround. Cap frame sizes and connection counts before parsing untrusted input.

Inspect secure client TLS defaults

import ssl

context = ssl.create_default_context()
print(context.check_hostname)
print(context.verify_mode == ssl.CERT_REQUIRED)

Output

True
True

Back to quick reference ↑

Local code tester

Build a bounded length-prefixed frame

Encode and parse an application frame without opening a network connection.

Runs in your browser

Python

import struct

def encode_frame(payload: bytes) -> bytes:
    if len(payload) > 1024:
        raise ValueError("frame too large")
    return struct.pack("!I", len(payload)) + payload

wire = encode_frame("Olá".encode("utf-8"))
size = struct.unpack("!I", wire[:4])[0]
print(size)
print(wire[4:4 + size].decode("utf-8"))

Output

Press Run to load Python locally.

Sources and further reading

References

Authoritative documentation used to verify and expand this cheat sheet.

Help us improve

Found a typo or missing example?

Tell us what would make this cheat sheet clearer, more complete, or more useful.

Share feedback

Quick reference

Detailed examples

Resolve addresses and connect with bounded waiting

Frame messages on TCP byte streams

Bind listeners without leaking accepted sockets

Distinguish deadlines from readiness

Drive many sockets with DefaultSelector

Preserve UDP datagram boundaries

Close protocols deliberately and add TLS

Build a bounded length-prefixed frame

More Python cheat sheets

Python asyncio and await Cheat Sheet

Python concurrent.futures and Parallel Work Cheat Sheet

Python Hashing, HMAC, and Secrets Cheat Sheet

References

Found a typo or missing example?