python
examples
examples.pyšpython
"""
Networking & APIs - Practical Examples
This file demonstrates HTTP requests, API interactions, and networking
"""
# ============================================================
# 1. URLLIB - BUILT-IN HTTP LIBRARY
# ============================================================
print("=" * 60)
print("1. URLLIB EXAMPLES")
print("=" * 60)
from urllib.request import urlopen, Request
from urllib.parse import urlparse, urlencode, parse_qs, urljoin
from urllib.error import HTTPError, URLError
import json
# Parse URLs
url = "https://api.github.com/users/python?tab=repos&page=1"
parsed = urlparse(url)
print(f"Scheme: {parsed.scheme}")
print(f"Host: {parsed.netloc}")
print(f"Path: {parsed.path}")
print(f"Query: {parsed.query}")
# Build query strings
params = {"q": "python", "sort": "stars", "order": "desc"}
query_string = urlencode(params)
print(f"Query string: {query_string}")
# Parse query strings
parsed_query = parse_qs("name=John&age=30&hobby=coding&hobby=reading")
print(f"Parsed query: {parsed_query}")
# Join URLs
base_url = "https://api.example.com/v1/"
endpoint = "users/123"
full_url = urljoin(base_url, endpoint)
print(f"Full URL: {full_url}")
# Make a simple request with urllib
try:
# Note: This is a real API call
with urlopen("https://api.github.com", timeout=10) as response:
print(f"\nStatus Code: {response.status}")
print(f"Headers: {dict(response.headers)}")
data = json.loads(response.read().decode())
print(f"API endpoints available: {len(data)} endpoints")
except (HTTPError, URLError) as e:
print(f"Request failed: {e}")
# ============================================================
# 2. REQUESTS LIBRARY - BASIC USAGE
# ============================================================
print("\n" + "=" * 60)
print("2. REQUESTS LIBRARY - BASIC USAGE")
print("=" * 60)
try:
import requests
REQUESTS_AVAILABLE = True
except ImportError:
print("Note: Install requests with 'pip install requests'")
REQUESTS_AVAILABLE = False
if REQUESTS_AVAILABLE:
# Simple GET request
response = requests.get("https://api.github.com")
print(f"Status: {response.status_code}")
print(f"OK: {response.ok}")
print(f"Headers: Content-Type = {response.headers.get('Content-Type')}")
# GET with parameters
response = requests.get(
"https://api.github.com/search/repositories",
params={"q": "python", "per_page": 5}
)
if response.ok:
data = response.json()
print(f"\nTotal Python repos found: {data.get('total_count', 0)}")
if data.get('items'):
print("Top 5 Python repositories:")
for repo in data['items'][:5]:
print(f" - {repo['full_name']}: {repo['stargazers_count']} stars")
# ============================================================
# 3. REQUESTS - DIFFERENT HTTP METHODS
# ============================================================
print("\n" + "=" * 60)
print("3. HTTP METHODS DEMONSTRATION")
print("=" * 60)
if REQUESTS_AVAILABLE:
# Using httpbin.org - a great testing API
base_url = "https://httpbin.org"
# GET request
print("\n--- GET Request ---")
response = requests.get(f"{base_url}/get", params={"name": "John", "age": 30})
print(f"GET args echoed: {response.json().get('args')}")
# POST with form data
print("\n--- POST (Form Data) ---")
response = requests.post(f"{base_url}/post", data={"username": "john", "email": "john@example.com"})
print(f"POST form data echoed: {response.json().get('form')}")
# POST with JSON
print("\n--- POST (JSON) ---")
response = requests.post(f"{base_url}/post", json={"product": "Widget", "price": 29.99})
print(f"POST JSON echoed: {response.json().get('json')}")
# PUT request
print("\n--- PUT Request ---")
response = requests.put(f"{base_url}/put", json={"id": 1, "name": "Updated"})
print(f"PUT JSON echoed: {response.json().get('json')}")
# DELETE request
print("\n--- DELETE Request ---")
response = requests.delete(f"{base_url}/delete")
print(f"DELETE successful: {response.ok}")
# Custom headers
print("\n--- Custom Headers ---")
headers = {
"User-Agent": "MyApp/1.0",
"Accept": "application/json",
"X-Custom-Header": "CustomValue"
}
response = requests.get(f"{base_url}/headers", headers=headers)
echoed_headers = response.json().get('headers', {})
print(f"Custom header echoed: X-Custom-Header = {echoed_headers.get('X-Custom-Header')}")
# ============================================================
# 4. JSON HANDLING
# ============================================================
print("\n" + "=" * 60)
print("4. JSON HANDLING")
print("=" * 60)
import json
from datetime import datetime, date
from decimal import Decimal
# Python to JSON
python_data = {
"name": "John Doe",
"age": 30,
"is_active": True,
"balance": 1000.50,
"hobbies": ["coding", "reading", "gaming"],
"address": {
"street": "123 Main St",
"city": "New York"
},
"friends": None
}
# Convert to JSON string
json_string = json.dumps(python_data)
print(f"JSON string (compact): {json_string[:60]}...")
# Pretty print
json_pretty = json.dumps(python_data, indent=2)
print(f"\nJSON string (pretty):\n{json_pretty}")
# Parse JSON string back to Python
parsed = json.loads(json_string)
print(f"\nParsed name: {parsed['name']}")
print(f"Parsed hobbies: {parsed['hobbies']}")
# Custom JSON encoder for non-serializable types
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, date):
return obj.strftime("%Y-%m-%d")
if isinstance(obj, Decimal):
return float(obj)
if isinstance(obj, set):
return list(obj)
return super().default(obj)
complex_data = {
"timestamp": datetime.now(),
"date": date.today(),
"price": Decimal("19.99"),
"tags": {"python", "api", "json"}
}
encoded = json.dumps(complex_data, cls=CustomEncoder, indent=2)
print(f"\nComplex data encoded:\n{encoded}")
# ============================================================
# 5. SESSIONS AND AUTHENTICATION
# ============================================================
print("\n" + "=" * 60)
print("5. SESSIONS AND AUTHENTICATION")
print("=" * 60)
if REQUESTS_AVAILABLE:
from requests.auth import HTTPBasicAuth
# Using sessions
print("--- Using Sessions ---")
session = requests.Session()
# Set default headers for all requests in this session
session.headers.update({
"User-Agent": "MyApp/1.0",
"Accept": "application/json"
})
# Session cookies persist across requests
response = session.get("https://httpbin.org/cookies/set/session_id/abc123")
response = session.get("https://httpbin.org/cookies")
print(f"Session cookies: {response.json().get('cookies')}")
# Basic Authentication
print("\n--- Basic Authentication ---")
response = requests.get(
"https://httpbin.org/basic-auth/user/passwd",
auth=HTTPBasicAuth("user", "passwd")
)
print(f"Auth successful: {response.ok}")
print(f"Auth response: {response.json()}")
# Bearer Token (simulated)
print("\n--- Bearer Token Authentication ---")
headers = {"Authorization": "Bearer my-secret-token"}
response = requests.get("https://httpbin.org/bearer", headers=headers)
print(f"Bearer auth successful: {response.ok}")
# ============================================================
# 6. ERROR HANDLING AND TIMEOUTS
# ============================================================
print("\n" + "=" * 60)
print("6. ERROR HANDLING AND TIMEOUTS")
print("=" * 60)
if REQUESTS_AVAILABLE:
from requests.exceptions import (
RequestException, Timeout, ConnectionError,
HTTPError, TooManyRedirects
)
# Timeout example
print("--- Timeout Handling ---")
try:
# 10 second delay, but 2 second timeout
response = requests.get("https://httpbin.org/delay/10", timeout=2)
except Timeout:
print("Request timed out (as expected)")
# HTTP error handling
print("\n--- HTTP Error Handling ---")
response = requests.get("https://httpbin.org/status/404")
print(f"404 Response: status={response.status_code}, ok={response.ok}")
try:
response.raise_for_status()
except HTTPError as e:
print(f"HTTPError caught: {e}")
# Connection error (will fail quickly with non-existent host)
print("\n--- Connection Error Handling ---")
try:
response = requests.get("http://definitely-not-a-real-website-xyz123.com", timeout=2)
except ConnectionError:
print("Connection failed (as expected)")
except Exception as e:
print(f"Other error: {type(e).__name__}")
# Generic error handling pattern
print("\n--- Robust Request Pattern ---")
def safe_request(url, **kwargs):
"""Make a request with comprehensive error handling."""
try:
response = requests.get(url, timeout=10, **kwargs)
response.raise_for_status()
return response
except Timeout:
print(f"Timeout: {url}")
except ConnectionError:
print(f"Connection failed: {url}")
except HTTPError as e:
print(f"HTTP error {e.response.status_code}: {url}")
except RequestException as e:
print(f"Request failed: {e}")
return None
result = safe_request("https://api.github.com")
if result:
print(f"Safe request succeeded: {result.status_code}")
# ============================================================
# 7. REAL API EXAMPLE - GITHUB API
# ============================================================
print("\n" + "=" * 60)
print("7. REAL API EXAMPLE - GITHUB API")
print("=" * 60)
if REQUESTS_AVAILABLE:
# Get user information
username = "python"
response = requests.get(f"https://api.github.com/users/{username}")
if response.ok:
user = response.json()
print(f"GitHub User: {user.get('login')}")
print(f" Name: {user.get('name')}")
print(f" Public Repos: {user.get('public_repos')}")
print(f" Followers: {user.get('followers')}")
print(f" Created: {user.get('created_at')}")
# Check rate limit
response = requests.get("https://api.github.com/rate_limit")
if response.ok:
rate = response.json()['rate']
print(f"\nRate Limit: {rate['remaining']}/{rate['limit']}")
# Get repository information
print("\nTop Python repositories:")
response = requests.get(
"https://api.github.com/search/repositories",
params={"q": "language:python", "sort": "stars", "per_page": 5}
)
if response.ok:
for repo in response.json().get('items', []):
print(f" ā {repo['stargazers_count']:,} - {repo['full_name']}")
# ============================================================
# 8. PAGINATION HANDLING
# ============================================================
print("\n" + "=" * 60)
print("8. PAGINATION HANDLING")
print("=" * 60)
if REQUESTS_AVAILABLE:
def get_paginated_data(url, params=None, max_pages=3):
"""Fetch paginated data from an API."""
if params is None:
params = {}
all_items = []
page = 1
while page <= max_pages:
params['page'] = page
params['per_page'] = 10
response = requests.get(url, params=params)
if not response.ok:
break
items = response.json()
if not items: # Empty page means we're done
break
all_items.extend(items)
print(f" Fetched page {page}: {len(items)} items")
page += 1
return all_items
# Fetch multiple pages of repos
print("Fetching Python org repositories (paginated):")
repos = get_paginated_data(
"https://api.github.com/orgs/python/repos",
max_pages=3
)
print(f"Total repos fetched: {len(repos)}")
# ============================================================
# 9. BUILDING A SIMPLE API CLIENT CLASS
# ============================================================
print("\n" + "=" * 60)
print("9. API CLIENT CLASS PATTERN")
print("=" * 60)
if REQUESTS_AVAILABLE:
class GitHubClient:
"""A simple GitHub API client."""
BASE_URL = "https://api.github.com"
def __init__(self, token=None):
self.session = requests.Session()
self.session.headers.update({
"Accept": "application/vnd.github.v3+json",
"User-Agent": "Python-GitHub-Client"
})
if token:
self.session.headers["Authorization"] = f"token {token}"
def _request(self, method, endpoint, **kwargs):
"""Make an API request."""
url = f"{self.BASE_URL}/{endpoint}"
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
return response.json()
def get_user(self, username):
"""Get user information."""
return self._request("GET", f"users/{username}")
def get_repos(self, username, page=1, per_page=10):
"""Get user repositories."""
params = {"page": page, "per_page": per_page}
return self._request("GET", f"users/{username}/repos", params=params)
def search_repos(self, query, sort="stars", order="desc"):
"""Search for repositories."""
params = {"q": query, "sort": sort, "order": order}
return self._request("GET", "search/repositories", params=params)
# Use the client
client = GitHubClient()
user = client.get_user("torvalds")
print(f"User: {user['login']} ({user['name']})")
print(f"Followers: {user['followers']:,}")
repos = client.get_repos("torvalds", per_page=3)
print(f"\nTop 3 repos:")
for repo in repos[:3]:
print(f" - {repo['name']}: {repo['stargazers_count']:,} stars")
# ============================================================
# 10. SOCKET PROGRAMMING EXAMPLE
# ============================================================
print("\n" + "=" * 60)
print("10. SOCKET PROGRAMMING")
print("=" * 60)
import socket
# Get host information
print("--- Host Information ---")
hostname = socket.gethostname()
print(f"Hostname: {hostname}")
# Resolve domain names
try:
github_ip = socket.gethostbyname("github.com")
print(f"github.com IP: {github_ip}")
except socket.gaierror:
print("Could not resolve github.com")
# Get address info
print("\n--- Address Information ---")
try:
addr_info = socket.getaddrinfo("google.com", 80, socket.AF_INET, socket.SOCK_STREAM)
for info in addr_info[:2]:
family, socktype, proto, canonname, sockaddr = info
print(f" {sockaddr[0]}:{sockaddr[1]}")
except socket.gaierror:
print("Could not get address info")
# Simple TCP client example (connecting to a web server)
print("\n--- Simple HTTP Request via Socket ---")
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)
sock.connect(("httpbin.org", 80))
# Send HTTP request
request = "GET /get HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n"
sock.send(request.encode())
# Receive response
response = b""
while True:
chunk = sock.recv(1024)
if not chunk:
break
response += chunk
sock.close()
# Parse response
response_text = response.decode()
status_line = response_text.split('\r\n')[0]
print(f"Response: {status_line}")
except socket.error as e:
print(f"Socket error: {e}")
# ============================================================
# SUMMARY
# ============================================================
print("\n" + "=" * 60)
print("NETWORKING & APIS SUMMARY")
print("=" * 60)
print("""
Key Concepts Covered:
ā urllib for basic HTTP requests
ā URL parsing and building
ā requests library for HTTP operations
ā Different HTTP methods (GET, POST, PUT, DELETE)
ā JSON serialization/deserialization
ā Custom JSON encoders
ā Sessions and authentication
ā Error handling and timeouts
ā Real API interactions (GitHub)
ā Pagination handling
ā API client class pattern
ā Basic socket programming
Best Practices:
1. Always use timeouts
2. Handle errors gracefully
3. Use sessions for multiple requests
4. Don't hardcode credentials
5. Implement retry logic for reliability
""")