Networking & APIs in Python

Overview

Python provides powerful tools for network communication and API interactions. This module covers everything from low-level sockets to high-level HTTP libraries.

---

1. HTTP Basics

What is HTTP?

HTTP (Hypertext Transfer Protocol) is the foundation of data communication on the web.

HTTP Methods

GET     - Retrieve data
POST    - Send data to create resource
PUT     - Update/replace resource
PATCH   - Partial update
DELETE  - Remove resource
HEAD    - Like GET but headers only
OPTIONS - Get supported methods

HTTP Status Codes

1xx - Informational
2xx - Success (200 OK, 201 Created, 204 No Content)
3xx - Redirection (301 Moved, 304 Not Modified)
4xx - Client Error (400 Bad Request, 401 Unauthorized, 404 Not Found)
5xx - Server Error (500 Internal Error, 503 Service Unavailable)

---

2. urllib - Built-in HTTP Library

urllib.request

from urllib.request import urlopen, Request
from urllib.parse import urlencode

# Simple GET request
response = urlopen('https://api.example.com/data')
data = response.read().decode('utf-8')

# With custom headers
req = Request('https://api.example.com/data')
req.add_header('Authorization', 'Bearer token123')
response = urlopen(req)

urllib.parse

from urllib.parse import urlparse, urlencode, parse_qs

# Parse URL
url = urlparse('https://example.com/path?key=value')
# ParseResult(scheme='https', netloc='example.com', path='/path', ...)

# Build query string
params = {'name': 'John', 'age': 30}
query = urlencode(params)  # 'name=John&age=30'

# Parse query string
parsed = parse_qs('name=John&age=30')
# {'name': ['John'], 'age': ['30']}

---

3. requests Library (Third-party)

The requests library is the most popular HTTP library in Python.

Installation

pip install requests

Basic Usage

import requests

# GET request
response = requests.get('https://api.github.com')
print(response.status_code)  # 200
print(response.headers)      # Response headers
print(response.text)         # Response body as string
print(response.json())       # Parse JSON response

# GET with parameters
params = {'q': 'python', 'page': 1}
response = requests.get('https://api.example.com/search', params=params)

# GET with headers
headers = {'Authorization': 'Bearer token123'}
response = requests.get('https://api.example.com/user', headers=headers)

POST Requests

import requests

# POST with form data
data = {'username': 'john', 'password': 'secret'}
response = requests.post('https://api.example.com/login', data=data)

# POST with JSON
import json
json_data = {'name': 'Product', 'price': 29.99}
response = requests.post(
    'https://api.example.com/products',
    json=json_data  # Automatically sets Content-Type
)

# File upload
files = {'file': open('document.pdf', 'rb')}
response = requests.post('https://api.example.com/upload', files=files)

Other HTTP Methods

# PUT
response = requests.put('https://api.example.com/user/1', json={'name': 'Jane'})

# PATCH
response = requests.patch('https://api.example.com/user/1', json={'email': 'new@email.com'})

# DELETE
response = requests.delete('https://api.example.com/user/1')

# HEAD
response = requests.head('https://api.example.com/file.zip')

Sessions

# Sessions persist cookies and connections
session = requests.Session()

# Login
session.post('https://api.example.com/login', data={'user': 'john', 'pass': 'secret'})

# Subsequent requests use same session (cookies preserved)
response = session.get('https://api.example.com/dashboard')

# Set default headers for all requests
session.headers.update({'Authorization': 'Bearer token123'})

Handling Responses

response = requests.get('https://api.example.com/data')

# Check for success
if response.ok:  # Status code < 400
    data = response.json()

# Raise exception for HTTP errors
response.raise_for_status()

# Response properties
response.status_code   # 200
response.headers       # {'Content-Type': 'application/json', ...}
response.text          # Response body as string
response.content       # Response body as bytes
response.json()        # Parse JSON
response.url           # Final URL (after redirects)
response.elapsed       # Time taken
response.encoding      # Response encoding

Timeouts and Error Handling

import requests
from requests.exceptions import RequestException, Timeout, ConnectionError

try:
    # Set timeout (connect, read)
    response = requests.get('https://api.example.com', timeout=(3, 10))
except Timeout:
    print("Request timed out")
except ConnectionError:
    print("Connection failed")
except RequestException as e:
    print(f"Request error: {e}")

---

4. JSON Handling

Working with JSON

import json

# Python object to JSON string
data = {'name': 'John', 'age': 30, 'hobbies': ['coding', 'reading']}
json_string = json.dumps(data)
# '{"name": "John", "age": 30, "hobbies": ["coding", "reading"]}'

# Pretty print
json_string = json.dumps(data, indent=2)

# JSON string to Python object
python_obj = json.loads(json_string)

# Read JSON from file
with open('data.json', 'r') as f:
    data = json.load(f)

# Write JSON to file
with open('data.json', 'w') as f:
    json.dump(data, f, indent=2)

Custom JSON Encoding

import json
from datetime import datetime

class DateTimeEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        return super().default(obj)

data = {'event': 'meeting', 'time': datetime.now()}
json_string = json.dumps(data, cls=DateTimeEncoder)

---

5. REST API Design Principles

REST Constraints

1. •Client-Server - Separation of concerns
2. •Stateless - Each request contains all needed info
3. •Cacheable - Responses must define cacheability
4. •Uniform Interface - Standard methods and URLs
5. •Layered System - Client can't tell if connected directly to server
6. •Code on Demand (optional) - Server can send executable code

URL Design

# Good REST URLs
GET    /users              # List all users
GET    /users/123          # Get user 123
POST   /users              # Create new user
PUT    /users/123          # Update user 123
DELETE /users/123          # Delete user 123

GET    /users/123/orders   # Get orders for user 123
POST   /users/123/orders   # Create order for user 123

---

6. Working with APIs

Authentication Methods

API Key

# In header
headers = {'X-API-Key': 'your-api-key'}
response = requests.get(url, headers=headers)

# In query parameter
params = {'api_key': 'your-api-key'}
response = requests.get(url, params=params)

Basic Auth

from requests.auth import HTTPBasicAuth

response = requests.get(url, auth=HTTPBasicAuth('username', 'password'))
# Or simply:
response = requests.get(url, auth=('username', 'password'))

OAuth 2.0 / Bearer Token

headers = {'Authorization': 'Bearer your-access-token'}
response = requests.get(url, headers=headers)

Pagination

def get_all_pages(base_url):
    all_results = []
    page = 1
    
    while True:
        response = requests.get(base_url, params={'page': page, 'per_page': 100})
        data = response.json()
        
        if not data['results']:
            break
            
        all_results.extend(data['results'])
        page += 1
        
    return all_results

Rate Limiting

import time
import requests

def rate_limited_request(url, max_retries=3):
    for attempt in range(max_retries):
        response = requests.get(url)
        
        if response.status_code == 429:  # Too Many Requests
            retry_after = int(response.headers.get('Retry-After', 60))
            print(f"Rate limited. Waiting {retry_after} seconds...")
            time.sleep(retry_after)
            continue
            
        return response
    
    raise Exception("Max retries exceeded")

---

7. Socket Programming

Low-level TCP Sockets

import socket

# Server
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(('localhost', 8000))
server.listen(5)

print("Server listening on port 8000")
while True:
    client, address = server.accept()
    data = client.recv(1024)
    print(f"Received: {data.decode()}")
    client.send(b"Hello from server!")
    client.close()

# Client
import socket

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('localhost', 8000))
client.send(b"Hello from client!")
response = client.recv(1024)
print(f"Response: {response.decode()}")
client.close()

UDP Sockets

import socket

# UDP Server
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server.bind(('localhost', 8000))

while True:
    data, addr = server.recvfrom(1024)
    print(f"From {addr}: {data.decode()}")
    server.sendto(b"ACK", addr)

---

8. Websockets

Using websockets Library

pip install websockets

import asyncio
import websockets

async def client():
    async with websockets.connect('ws://localhost:8765') as websocket:
        await websocket.send("Hello!")
        response = await websocket.recv()
        print(f"Received: {response}")

asyncio.run(client())

Websocket Server

import asyncio
import websockets

async def handler(websocket, path):
    async for message in websocket:
        print(f"Received: {message}")
        await websocket.send(f"Echo: {message}")

async def main():
    async with websockets.serve(handler, "localhost", 8765):
        await asyncio.Future()  # Run forever

asyncio.run(main())

---

9. httpx - Modern HTTP Client

Installation

pip install httpx

Basic Usage

import httpx

# Sync client
response = httpx.get('https://api.example.com')
print(response.json())

# Async client
async def fetch():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com')
        return response.json()

HTTP/2 Support

async with httpx.AsyncClient(http2=True) as client:
    response = await client.get('https://api.example.com')

---

10. Best Practices

1. Always Use Timeouts

response = requests.get(url, timeout=10)

2. Use Sessions for Multiple Requests

with requests.Session() as session:
    for url in urls:
        response = session.get(url)

3. Handle Errors Gracefully

try:
    response = requests.get(url, timeout=10)
    response.raise_for_status()
except requests.RequestException as e:
    logger.error(f"Request failed: {e}")

4. Don't Hardcode Credentials

import os
API_KEY = os.environ.get('API_KEY')

5. Use Connection Pooling

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
retries = Retry(total=3, backoff_factor=0.1)
adapter = HTTPAdapter(max_retries=retries, pool_connections=10, pool_maxsize=10)
session.mount('http://', adapter)
session.mount('https://', adapter)

---

Summary

---

Next Steps

After mastering networking and APIs:

1. •Build a REST API client for a real service
2. •Create a simple chat application with websockets
3. •Learn about API rate limiting and caching strategies
4. •Explore GraphQL APIs