Python socket: Low-Level Networking

import socket
 
# Create TCP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
# Connect to server
sock.connect(('example.com', 80))
 
# Send HTTP request
request = b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n'
sock.sendall(request)
 
# Receive response
response = sock.recv(4096)
print(response.decode())
 
sock.close()

import socket
 
# Create server socket
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
# Bind and listen
server.bind(('localhost', 8080))
server.listen(5)
 
print("Server listening on port 8080")
 
while True:
    # Accept connection
    client, address = server.accept()
    print(f"Connection from {address}")
    
    # Handle client
    data = client.recv(1024)
    client.sendall(b'Hello from server!')
    client.close()

import socket
 
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
    sock.connect(('example.com', 80))
    sock.sendall(b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n')
    response = sock.recv(4096)
# Socket automatically closed

import socket
 
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 
# Send datagram (no connection needed)
sock.sendto(b'Hello, UDP!', ('localhost', 9999))
 
# Receive response
data, addr = sock.recvfrom(1024)
print(f"Received {data} from {addr}")
 
sock.close()

import socket
 
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind(('localhost', 9999))
 
print("UDP server listening on port 9999")
 
while True:
    data, addr = sock.recvfrom(1024)
    print(f"Received {data} from {addr}")
    sock.sendto(b'ACK', addr)

import socket
 
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setblocking(False)
 
try:
    sock.connect(('example.com', 80))
except BlockingIOError:
    pass  # Connection in progress
 
# Later, check if ready
sock.setblocking(True)
sock.sendall(b'GET / HTTP/1.1\r\n\r\n')

import socket
 
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5.0)  # 5 second timeout
 
try:
    sock.connect(('example.com', 80))
    sock.sendall(b'GET / HTTP/1.1\r\n\r\n')
    response = sock.recv(4096)
except socket.timeout:
    print("Operation timed out")

import socket
 
# Get IP address
ip = socket.gethostbyname('example.com')
print(f"IP: {ip}")
 
# Get full info
info = socket.getaddrinfo('example.com', 80)
for family, socktype, proto, canonname, sockaddr in info:
    print(f"{sockaddr}")
 
# Reverse lookup
hostname = socket.gethostbyaddr('93.184.216.34')
print(f"Hostname: {hostname[0]}")

import socket
 
# Local hostname
hostname = socket.gethostname()
print(f"Hostname: {hostname}")
 
# Local IP
local_ip = socket.gethostbyname(hostname)
print(f"Local IP: {local_ip}")
 
# Get all local IPs
for info in socket.getaddrinfo(hostname, None):
    print(info[4][0])

import socket
 
def echo_server(port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind(('', port))
        server.listen(1)
        
        print(f"Echo server on port {port}")
        
        while True:
            conn, addr = server.accept()
            with conn:
                print(f"Connected: {addr}")
                while True:
                    data = conn.recv(1024)
                    if not data:
                        break
                    conn.sendall(data)
 
echo_server(8888)

import socket
import threading
 
def handle_client(conn, addr):
    print(f"Handling {addr}")
    with conn:
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(data.upper())
    print(f"Disconnected: {addr}")
 
def threaded_server(port):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind(('', port))
        server.listen(5)
        
        while True:
            conn, addr = server.accept()
            thread = threading.Thread(target=handle_client, args=(conn, addr))
            thread.start()

import socket
import select
 
def select_server(port):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.setblocking(False)
    server.bind(('', port))
    server.listen(5)
    
    inputs = [server]
    
    while inputs:
        readable, _, _ = select.select(inputs, [], [])
        
        for sock in readable:
            if sock is server:
                conn, addr = server.accept()
                conn.setblocking(False)
                inputs.append(conn)
            else:
                data = sock.recv(1024)
                if data:
                    sock.sendall(data)
                else:
                    inputs.remove(sock)
                    sock.close()

import socket
 
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
# Reuse address immediately after close
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
# Keep connection alive
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
 
# Set buffer sizes
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8192)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 8192)
 
# Disable Nagle's algorithm (send immediately)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

import socket
import os
 
# Server
server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
socket_path = '/tmp/mysocket'
 
if os.path.exists(socket_path):
    os.remove(socket_path)
 
server.bind(socket_path)
server.listen(1)
 
# Client
client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
client.connect(socket_path)