9 Advanced Python Techniques for Efficient API Integration

import httpx # Synchronous request def get_user_sync(user_id): response = httpx.get(f"https://api.example.com/users/{user_id}") response.raise_for_status() return response.json() # Asynchronous request async def get_user_async(user_id): async with httpx.AsyncClient() as client: response = await client.get(f"https://api.example.com/users/{user_id}") response.raise_for_status() return response.json() 

import aiohttp import asyncio async def fetch_multiple_users(user_ids): async with aiohttp.ClientSession() as session: tasks = [fetch_user(session, user_id) for user_id in user_ids] return await asyncio.gather(*tasks) async def fetch_user(session, user_id): url = f"https://api.example.com/users/{user_id}" async with session.get(url) as response: return await response.json() 

from pydantic import BaseModel, Field, validator from typing import List, Optional from datetime import datetime class User(BaseModel): id: int name: str email: str created_at: datetime profile_image: Optional[str] = None @validator('email') def email_must_be_valid(cls, v): if '@' not in v: raise ValueError('Invalid email format') return v async def get_validated_user(user_id): async with httpx.AsyncClient() as client: response = await client.get(f"https://api.example.com/users/{user_id}") response.raise_for_status() # Automatic validation and type conversion  return User(**response.json()) 

from functools import lru_cache from cachetools import TTLCache import time # In-memory cache with TTL user_cache = TTLCache(maxsize=100, ttl=300) # 5 minute TTL  def get_user(user_id): cache_key = f"user:{user_id}" # Check cache  if cache_key in user_cache: return user_cache[cache_key] # Fetch from API  response = httpx.get(f"https://api.example.com/users/{user_id}") response.raise_for_status() data = response.json() # Update cache  user_cache[cache_key] = data return data # For immutable data, we can use lru_cache @lru_cache(maxsize=128) def get_country_data(country_code): response = httpx.get(f"https://api.example.com/countries/{country_code}") response.raise_for_status() return response.json() 

import redis import json redis_client = redis.Redis(host='localhost', port=6379, db=0) def get_cached_data(key, fetch_function, ttl=300): # Try to get from cache  cached = redis_client.get(key) if cached: return json.loads(cached) # Fetch fresh data  data = fetch_function() # Store in cache  redis_client.setex(key, ttl, json.dumps(data)) return data def fetch_weather_data(city): return get_cached_data( f"weather:{city}", lambda: httpx.get(f"https://api.weather.com/{city}").json(), ttl=1800 # 30 minutes  ) 

import time import random from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type class RateLimitExceeded(Exception): pass @retry( wait=wait_exponential(multiplier=1, min=2, max=60), stop=stop_after_attempt(5), retry=retry_if_exception_type(RateLimitExceeded) ) def get_user_with_retry(user_id): response = httpx.get(f"https://api.example.com/users/{user_id}") if response.status_code == 429: retry_after = int(response.headers.get('Retry-After', 5)) # Add jitter to avoid thundering herd  jitter = random.uniform(0, 1) time.sleep(retry_after + jitter) raise RateLimitExceeded("Rate limit exceeded") response.raise_for_status() return response.json() 

import time class TokenBucket: def __init__(self, tokens, fill_rate): self.capacity = tokens self.tokens = tokens self.fill_rate = fill_rate self.timestamp = time.time() def consume(self, tokens=1): # Update token count  now = time.time() elapsed = now - self.timestamp self.tokens = min(self.capacity, self.tokens + elapsed * self.fill_rate) self.timestamp = now # Check if enough tokens  if tokens <= self.tokens: self.tokens -= tokens return True return False # Usage rate_limiter = TokenBucket(tokens=60, fill_rate=1) # 60 requests per minute  def call_api(endpoint): if rate_limiter.consume(): return httpx.get(f"https://api.example.com/{endpoint}") else: time.sleep(1) # Wait a bit  return call_api(endpoint) # Try again 

import asyncio from typing import List, Dict, Any, AsyncGenerator async def paginate_all_results(endpoint: str) -> List[Dict[Any, Any]]: all_results = [] page = 1 while True: async with httpx.AsyncClient() as client: response = await client.get( f"https://api.example.com/{endpoint}", params={"page": page, "per_page": 100} ) response.raise_for_status() data = response.json() if not data: break all_results.extend(data) # Check if we've reached the last page  if len(data) < 100: break page += 1 return all_results # For memory-efficient processing of large datasets async def stream_paginated_results(endpoint: str) -> AsyncGenerator[Dict[Any, Any], None]: page = 1 while True: async with httpx.AsyncClient() as client: response = await client.get( f"https://api.example.com/{endpoint}", params={"page": page, "per_page": 100} ) response.raise_for_status() page_data = response.json() if not page_data: break # Yield individual items  for item in page_data: yield item # Check if we've reached the last page  if len(page_data) < 100: break page += 1 

import os import jwt from datetime import datetime, timedelta from dotenv import load_dotenv load_dotenv() class TokenManager: def __init__(self): self.api_key = os.getenv("API_KEY") self.api_secret = os.getenv("API_SECRET") self.token = None self.token_expiry = None def get_valid_token(self): # Check if token exists and is still valid  if self.token and self.token_expiry and datetime.now() < self.token_expiry: return self.token # Generate new token  self.token = self._generate_token() self.token_expiry = datetime.now() + timedelta(hours=1) return self.token def _generate_token(self): payload = { "iss": self.api_key, "exp": datetime.now() + timedelta(hours=1), "iat": datetime.now() } return jwt.encode(payload, self.api_secret, algorithm="HS256") # Usage token_manager = TokenManager() def call_protected_api(endpoint): token = token_manager.get_valid_token() headers = {"Authorization": f"Bearer {token}"} return httpx.get(f"https://api.example.com/{endpoint}", headers=headers) 

import time from httpx import Client class OAuth2Client: def __init__(self, client_id, client_secret, token_url): self.client_id = client_id self.client_secret = client_secret self.token_url = token_url self.access_token = None self.refresh_token = None self.expires_at = 0 def get_headers(self): if not self.access_token or time.time() > self.expires_at - 60: self._refresh_token() return {"Authorization": f"Bearer {self.access_token}"} def _refresh_token(self): with Client() as client: data = { "grant_type": "refresh_token" if self.refresh_token else "client_credentials", "client_id": self.client_id, "client_secret": self.client_secret, } if self.refresh_token: data["refresh_token"] = self.refresh_token response = client.post(self.token_url, data=data) response.raise_for_status() token_data = response.json() self.access_token = token_data["access_token"] self.refresh_token = token_data.get("refresh_token", self.refresh_token) self.expires_at = time.time() + token_data.get("expires_in", 3600) 

import time from enum import Enum class CircuitState(Enum): CLOSED = 1 # Normal operation  OPEN = 2 # Failing, don't try  HALF_OPEN = 3 # Testing if working again  class CircuitBreaker: def __init__(self, failure_threshold=5, recovery_timeout=30, timeout=10): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.timeout = timeout self.state = CircuitState.CLOSED self.failures = 0 self.last_failure_time = 0 def __call__(self, func): def wrapper(*args, **kwargs): if self.state == CircuitState.OPEN: if time.time() > self.last_failure_time + self.recovery_timeout: self.state = CircuitState.HALF_OPEN else: raise Exception("Circuit breaker is open") try: result = func(*args, **kwargs) # Reset on success  if self.state == CircuitState.HALF_OPEN: self.failures = 0 self.state = CircuitState.CLOSED return result except Exception as e: self.failures += 1 self.last_failure_time = time.time() if self.failures >= self.failure_threshold or self.state == CircuitState.HALF_OPEN: self.state = CircuitState.OPEN raise e return wrapper # Usage @CircuitBreaker(failure_threshold=3, recovery_timeout=60) def call_potentially_failing_api(): return httpx.get("https://api.example.com/endpoint", timeout=5.0) 

# Install with: pip install openapi-python-client # Then generate with: openapi-python-client generate --url https://api.example.com/openapi.json  # Example usage of a generated client from example_client import Client from example_client.api.users import get_user, create_user from example_client.models import User, UserCreate client = Client(base_url="https://api.example.com", token="your-token") # Get a user user_response = get_user.sync(client=client, user_id=123) user = user_response.parsed # Create a user new_user = UserCreate(name="John Doe", email="john@example.com") create_response = create_user.sync(client=client, json_body=new_user) 

from gql import Client, gql from gql.transport.aiohttp import AIOHTTPTransport async def fetch_user_data(user_id): transport = AIOHTTPTransport(url="https://api.example.com/graphql") async with Client(transport=transport) as client: query = gql(""" query GetUser($id: ID!) { user(id: $id) { id name email posts { id title } } } """) variables = {"id": user_id} result = await client.execute(query, variable_values=variables) return result 

import time import statistics from dataclasses import dataclass, field from typing import List, Dict @dataclass class APIMetrics: endpoint: str response_times: List[float] = field(default_factory=list) status_counts: Dict[int, int] = field(default_factory=dict) error_count: int = 0 def add_response(self, status_code, response_time): self.response_times.append(response_time) self.status_counts[status_code] = self.status_counts.get(status_code, 0) + 1 if status_code >= 400: self.error_count += 1 @property def avg_response_time(self): if not self.response_times: return 0 return statistics.mean(self.response_times) @property def p95_response_time(self): if not self.response_times: return 0 return statistics.quantiles(self.response_times, n=20)[19] # 95th percentile  @property def success_rate(self): total = sum(self.status_counts.values()) if total == 0: return 1.0 return 1 - (self.error_count / total) # Metrics collection metrics = {} def track_api_call(endpoint): def decorator(func): def wrapper(*args, **kwargs): if endpoint not in metrics: metrics[endpoint] = APIMetrics(endpoint=endpoint) start_time = time.time() try: response = func(*args, **kwargs) elapsed = time.time() - start_time metrics[endpoint].add_response(response.status_code, elapsed) return response except Exception as e: elapsed = time.time() - start_time metrics[endpoint].add_response(500, elapsed) raise e return wrapper return decorator # Usage @track_api_call("get_user") def get_user(user_id): return httpx.get(f"https://api.example.com/users/{user_id}")