Foundations of Python Network Programming Skill

Note: This book uses Python 2. Examples updated to Python 3 patterns where relevant.

TCP/UDP Socket Fundamentals

TCP Client

import socket

def tcp_client(host, port, data):
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((host, port))
    s.sendall(data.encode())

    chunks = []
    while True:
        chunk = s.recv(4096)
        if not chunk:
            break
        chunks.append(chunk)
    s.close()
    return b''.join(chunks)

UDP Client

import socket

def udp_client(host, port, data):
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.settimeout(5)                          # UDP has no connection state
    s.sendto(data.encode(), (host, port))
    try:
        response, addr = s.recvfrom(65535)
        return response
    except socket.timeout:
        return None
    finally:
        s.close()

Socket Timeouts

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(10)                             # 10 second timeout
try:
    s.connect((host, port))
except socket.timeout:
    print("Connection timed out")

# Non-blocking with select
import select
readable, writable, exceptional = select.select([s], [s], [s], timeout=5)

Network Byte Order

import struct

# Pack/unpack integers for network transmission (big-endian = network order)
data = struct.pack('!I', 12345)              # '!' = network byte order, I = unsigned int
value = struct.unpack('!I', data)[0]

# Send a length-prefixed message (framing protocol)
def send_message(sock, msg):
    data = msg.encode()
    length = struct.pack('!I', len(data))
    sock.sendall(length + data)

def recv_message(sock):
    raw_len = recv_exact(sock, 4)
    msglen = struct.unpack('!I', raw_len)[0]
    return recv_exact(sock, msglen).decode()

def recv_exact(sock, n):
    data = b''
    while len(data) < n:
        packet = sock.recv(n - len(data))
        if not packet:
            raise ConnectionError("Connection closed")
        data += packet
    return data

---

TCP Server Patterns

Simple Iterative Server

import socket

def tcp_server(host, port, handler):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    print(f"Listening on {host}:{port}")

    while True:
        conn, addr = server.accept()
        try:
            handler(conn, addr)
        finally:
            conn.close()

select()-Based Multiplexed Server (I/O without threads)

import select, socket

def select_server(host, port):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    server.setblocking(False)

    inputs = [server]

    while True:
        readable, _, _ = select.select(inputs, [], [])
        for s in readable:
            if s is server:
                conn, addr = s.accept()
                conn.setblocking(False)
                inputs.append(conn)
            else:
                data = s.recv(4096)
                if data:
                    s.sendall(data)          # echo back
                else:
                    inputs.remove(s)
                    s.close()

---

DNS Queries

import socket

# Basic: OS resolver
ip = socket.gethostbyname('example.com')
name, aliases, addresses = socket.gethostbyaddr('93.184.216.34')  # reverse lookup

# Advanced: dnspython (modern replacement for PyDNS)
import dns.resolver

# A record
answers = dns.resolver.resolve('example.com', 'A')
for rdata in answers:
    print(rdata.address)

# MX record (mail servers)
answers = dns.resolver.resolve('example.com', 'MX')
for rdata in answers:
    print(f"{rdata.preference} {rdata.exchange}")

# TXT record
answers = dns.resolver.resolve('example.com', 'TXT')
for rdata in answers:
    print(str(rdata))

# PTR (reverse)
answers = dns.resolver.resolve(dns.reversename.from_address('8.8.8.8'), 'PTR')

---

HTTP/Web Client

import urllib.request, urllib.parse

# Simple GET
with urllib.request.urlopen('http://example.com') as resp:
    content = resp.read().decode()

# POST form data
data = urllib.parse.urlencode({'username': 'user', 'password': 'pass'}).encode()
req = urllib.request.Request('http://example.com/login', data=data)
with urllib.request.urlopen(req) as resp:
    content = resp.read()

# With authentication
password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()
password_mgr.add_password(None, 'http://example.com/', 'user', 'pass')
auth_handler = urllib.request.HTTPBasicAuthHandler(password_mgr)
opener = urllib.request.build_opener(auth_handler)
urllib.request.install_opener(opener)

# Modern alternative: requests library
import requests
r = requests.get('http://example.com', auth=('user', 'pass'))
r = requests.post('http://example.com/login', data={'user': 'u', 'pass': 'p'})

HTML Parsing

from html.parser import HTMLParser

class LinkParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.links = []

    def handle_starttag(self, tag, attrs):
        if tag == 'a':
            for attr, value in attrs:
                if attr == 'href' and value:
                    self.links.append(value)

parser = LinkParser()
parser.feed(html_content)
print(parser.links)

# Modern: BeautifulSoup
from bs4 import BeautifulSoup
soup = BeautifulSoup(html_content, 'html.parser')
links = [a['href'] for a in soup.find_all('a', href=True)]
text = soup.find('div', class_='content').get_text()

---

Email

Send Email (SMTP)

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_email(smtp_host, smtp_port, username, password, to, subject, body):
    msg = MIMEMultipart()
    msg['From'] = username
    msg['To'] = to
    msg['Subject'] = subject
    msg.attach(MIMEText(body, 'plain'))

    with smtplib.SMTP(smtp_host, smtp_port) as server:
        server.ehlo()
        server.starttls()                    # STARTTLS for encryption
        server.login(username, password)
        server.sendmail(username, to, msg.as_string())

# With attachment
from email.mime.base import MIMEBase
from email import encoders

attachment = MIMEBase('application', 'octet-stream')
with open('file.txt', 'rb') as f:
    attachment.set_payload(f.read())
encoders.encode_base64(attachment)
attachment.add_header('Content-Disposition', 'attachment; filename=file.txt')
msg.attach(attachment)

POP3 (Retrieve Email)

import poplib
from email import message_from_bytes

def get_pop3_messages(host, username, password):
    server = poplib.POP3_SSL(host)
    server.user(username)
    server.pass_(password)

    num_messages = len(server.list()[1])
    messages = []
    for i in range(num_messages):
        raw_msg = b'\n'.join(server.retr(i + 1)[1])
        msg = message_from_bytes(raw_msg)
        messages.append(msg)

    server.quit()
    return messages

IMAP (Full Mailbox Access)

import imaplib
from email import message_from_bytes

def imap_search(host, username, password, criteria='ALL'):
    server = imaplib.IMAP4_SSL(host)
    server.login(username, password)
    server.select('INBOX')

    _, msg_ids = server.search(None, criteria)
    messages = []
    for msg_id in msg_ids[0].split():
        _, msg_data = server.fetch(msg_id, '(RFC822)')
        msg = message_from_bytes(msg_data[0][1])
        messages.append(msg)

    server.logout()
    return messages

# IMAP search criteria examples:
# 'UNSEEN'          → unread messages
# 'FROM "sender"'   → from specific sender
# 'SUBJECT "test"'  → subject match
# 'SINCE 01-Jan-2024' → date filter

---

FTP

from ftplib import FTP, FTP_TLS

def ftp_download(host, username, password, remote_file, local_file):
    with FTP(host) as ftp:
        ftp.login(username, password)
        ftp.cwd('/pub')                      # change directory

        with open(local_file, 'wb') as f:
            ftp.retrbinary(f'RETR {remote_file}', f.write)

def ftp_upload(host, username, password, local_file, remote_name):
    with FTP(host) as ftp:
        ftp.login(username, password)
        with open(local_file, 'rb') as f:
            ftp.storbinary(f'STOR {remote_name}', f)

def ftp_list_recursive(ftp, path='/'):
    """Recursively list FTP directory."""
    items = []
    try:
        ftp.cwd(path)
        names = ftp.nlst()
        for name in names:
            try:
                ftp.cwd(name)
                items.extend(ftp_list_recursive(ftp, f'{path}/{name}'))
                ftp.cwd('..')
            except:
                items.append(f'{path}/{name}')  # it's a file
    except:
        pass
    return items

---

SSL/TLS

import ssl, socket

# SSL client
context = ssl.create_default_context()       # verify server cert
context.load_verify_locations('/etc/ssl/certs/ca-certificates.crt')

with socket.create_connection((host, port)) as raw_sock:
    with context.wrap_socket(raw_sock, server_hostname=host) as ssl_sock:
        ssl_sock.sendall(b'GET / HTTP/1.0\r\n\r\n')
        data = ssl_sock.read()

# SSL server
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain('server.crt', 'server.key')

with socket.socket() as raw_sock:
    raw_sock.bind((host, port))
    raw_sock.listen(1)
    with context.wrap_socket(raw_sock, server_side=True) as ssl_sock:
        conn, addr = ssl_sock.accept()

---

Database Clients (DB-API 2.0)

import sqlite3    # or psycopg2 for PostgreSQL, mysql.connector for MySQL

conn = sqlite3.connect('database.db')   # or connect(host, user, pass, db)
cursor = conn.cursor()

# DDL
cursor.execute('''CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT, email TEXT)''')
conn.commit()

# Parameterized queries (prevent SQL injection)
cursor.execute("INSERT INTO users (name, email) VALUES (?, ?)", (name, email))
cursor.execute("SELECT * FROM users WHERE email = ?", (email,))
rows = cursor.fetchall()

# Bulk operations
cursor.executemany("INSERT INTO users VALUES (?, ?, ?)", records_list)

# Context manager
with conn:                               # auto-commit on success, rollback on exception
    cursor.execute("UPDATE users SET name = ? WHERE id = ?", (name, uid))

---

Concurrency Patterns

Forking Server

import socket, os

def forking_server(host, port, handler):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(10)

    while True:
        conn, addr = server.accept()
        pid = os.fork()
        if pid == 0:                         # child process
            server.close()
            handler(conn, addr)
            conn.close()
            os._exit(0)
        else:                                # parent process
            conn.close()
            # clean up zombie children
            os.waitpid(-1, os.WNOHANG)

Threading Server

import socket, threading

def threading_server(host, port, handler):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(10)

    while True:
        conn, addr = server.accept()
        t = threading.Thread(target=handler, args=(conn, addr))
        t.daemon = True
        t.start()

# Thread pool (better for high concurrency)
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=20)
while True:
    conn, addr = server.accept()
    executor.submit(handler, conn, addr)

Async I/O (modern asyncio — not in original book)

import asyncio

async def handle_client(reader, writer):
    data = await reader.read(4096)
    writer.write(data)                       # echo
    await writer.drain()
    writer.close()

async def main():
    server = await asyncio.start_server(handle_client, '0.0.0.0', 8888)
    async with server:
        await server.serve_forever()

asyncio.run(main())

---

SocketServer Framework (Built-in)

from socketserver import TCPServer, BaseRequestHandler, ThreadingMixIn

class MyHandler(BaseRequestHandler):
    def handle(self):
        data = self.request.recv(1024)
        self.request.sendall(data.upper())   # echo uppercase

class ThreadedServer(ThreadingMixIn, TCPServer):
    allow_reuse_address = True

server = ThreadedServer(('0.0.0.0', 9999), MyHandler)
server.serve_forever()

---

Key Patterns Reference

Task	Modern Python Tool
HTTP requests	requests or httpx
WebSocket	websockets
Async networking	asyncio + aiohttp
SSL/TLS	ssl module + context
DNS queries	dnspython
IMAP	imaplib (stdlib) or imapclient
SMTP	smtplib (stdlib)
FTP	ftplib (stdlib)
Database	psycopg2, aiomysql, sqlalchemy
Struct packing	struct module