On this tutorial, we information customers by constructing a strong, production-ready Python SDK. It begins by exhibiting easy methods to set up and configure important asynchronous HTTP libraries (aiohttp, nest-asyncio). It then walks by the implementation of core elements, together with structured response objects, token-bucket charge limiting, in-memory caching with TTL, and a clear, dataclass-driven design. We’ll see easy methods to wrap these items up in an AdvancedSDK class that helps async context administration, automated retry/wait-on-rate-limit habits, JSON/auth headers injection, and handy HTTP-verb strategies. Alongside the best way, a demo harness towards JSONPlaceholder illustrates caching effectivity, batch fetching with charge limits, error dealing with, and even exhibits easy methods to prolong the SDK through a fluent “builder” sample for customized configuration.
import asyncio
import aiohttp
import time
import json
from typing import Dict, Checklist, Elective, Any, Union
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
import hashlib
import logging
!pip set up aiohttp nest-asyncioWe set up and configure the asynchronous runtime by importing asyncio and aiohttp, alongside utilities for timing, JSON dealing with, dataclass modeling, caching (through hashlib and datetime), and structured logging. The !pip set up aiohttp nest-asyncio line ensures that the pocket book can run an occasion loop seamlessly inside Colab, enabling sturdy async HTTP requests and rate-limited workflows.
@dataclass
class APIResponse:
"""Structured response object"""
knowledge: Any
status_code: int
headers: Dict[str, str]
timestamp: datetime
def to_dict(self) -> Dict:
return asdict(self)The APIResponse dataclass encapsulates HTTP response particulars, payload (knowledge), standing code, headers, and the timestamp of retrieval right into a single, typed object. The to_dict() helper converts the occasion right into a plain dictionary for simple logging, serialization, or downstream processing.
class RateLimiter:
"""Token bucket charge limiter"""
def __init__(self, max_calls: int = 100, time_window: int = 60):
self.max_calls = max_calls
self.time_window = time_window
self.calls = []
def can_proceed(self) -> bool:
now = time.time()
self.calls = [call_time for call_time in self.calls if now - call_time < self.time_window]
if len(self.calls) < self.max_calls:
self.calls.append(now)
return True
return False
def wait_time(self) -> float:
if not self.calls:
return 0
return max(0, self.time_window - (time.time() - self.calls[0]))The RateLimiter class enforces a easy token-bucket coverage by monitoring the timestamps of latest calls and permitting as much as max_calls inside a rolling time_window. When the restrict is reached, can_proceed() returns False, and wait_time() calculates how lengthy to pause earlier than making the following request.
class Cache:
"""Easy in-memory cache with TTL"""
def __init__(self, default_ttl: int = 300):
self.cache = {}
self.default_ttl = default_ttl
def _generate_key(self, technique: str, url: str, params: Dict = None) -> str:
key_data = f"{technique}:{url}:{json.dumps(params or {}, sort_keys=True)}"
return hashlib.md5(key_data.encode()).hexdigest()
def get(self, technique: str, url: str, params: Dict = None) -> Elective[APIResponse]:
key = self._generate_key(technique, url, params)
if key in self.cache:
response, expiry = self.cache[key]
if datetime.now() < expiry:
return response
del self.cache[key]
return None
def set(self, technique: str, url: str, response: APIResponse, params: Dict = None, ttl: int = None):
key = self._generate_key(technique, url, params)
expiry = datetime.now() + timedelta(seconds=ttl or self.default_ttl)
self.cache[key] = (response, expiry)The Cache class offers a light-weight in-memory TTL cache for API responses by hashing the request signature (technique, URL, params) into a singular key. It returns legitimate cached APIResponse objects earlier than expiry and robotically evicts stale entries after their time-to-live has elapsed.
class AdvancedSDK:
"""Superior SDK with fashionable Python patterns"""
def __init__(self, base_url: str, api_key: str = None, rate_limit: int = 100):
self.base_url = base_url.rstrip('/')
self.api_key = api_key
self.session = None
self.rate_limiter = RateLimiter(max_calls=rate_limit)
self.cache = Cache()
self.logger = self._setup_logger()
def _setup_logger(self) -> logging.Logger:
logger = logging.getLogger(f"SDK-{id(self)}")
if not logger.handlers:
handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s - %(title)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
return logger
async def __aenter__(self):
"""Async context supervisor entry"""
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context supervisor exit"""
if self.session:
await self.session.shut()
def _get_headers(self) -> Dict[str, str]:
headers = {'Content material-Kind': 'utility/json'}
if self.api_key:
headers['Authorization'] = f'Bearer {self.api_key}'
return headers
async def _make_request(self, technique: str, endpoint: str, params: Dict = None,
knowledge: Dict = None, use_cache: bool = True) -> APIResponse:
"""Core request technique with charge limiting and caching"""
if use_cache and technique.higher() == 'GET':
cached = self.cache.get(technique, endpoint, params)
if cached:
self.logger.data(f"Cache hit for {technique} {endpoint}")
return cached
if not self.rate_limiter.can_proceed():
wait_time = self.rate_limiter.wait_time()
self.logger.warning(f"Charge restrict hit, ready {wait_time:.2f}s")
await asyncio.sleep(wait_time)
url = f"{self.base_url}/{endpoint.lstrip('/')}"
attempt:
async with self.session.request(
technique=technique.higher(),
url=url,
params=params,
json=knowledge,
headers=self._get_headers()
) as resp:
response_data = await resp.json() if resp.content_type == 'utility/json' else await resp.textual content()
api_response = APIResponse(
knowledge=response_data,
status_code=resp.standing,
headers=dict(resp.headers),
timestamp=datetime.now()
)
if use_cache and technique.higher() == 'GET' and 200 <= resp.standing < 300:
self.cache.set(technique, endpoint, api_response, params)
self.logger.data(f"{technique.higher()} {endpoint} - Standing: {resp.standing}")
return api_response
besides Exception as e:
self.logger.error(f"Request failed: {str(e)}")
increase
async def get(self, endpoint: str, params: Dict = None, use_cache: bool = True) -> APIResponse:
return await self._make_request('GET', endpoint, params=params, use_cache=use_cache)
async def put up(self, endpoint: str, knowledge: Dict = None) -> APIResponse:
return await self._make_request('POST', endpoint, knowledge=knowledge, use_cache=False)
async def put(self, endpoint: str, knowledge: Dict = None) -> APIResponse:
return await self._make_request('PUT', endpoint, knowledge=knowledge, use_cache=False)
async def delete(self, endpoint: str) -> APIResponse:
return await self._make_request('DELETE', endpoint, use_cache=False)The AdvancedSDK class wraps every thing collectively right into a clear, async-first consumer: it manages an aiohttp session through async context managers, injects JSON and auth headers, and coordinates our RateLimiter and Cache underneath the hood. Its _make_request technique centralizes GET/POST/PUT/DELETE logic, dealing with cache lookups, rate-limit waits, error logging, and response packing into APIResponse objects, whereas the get/put up/put/delete helpers give us ergonomic, high-level calls.
async def demo_sdk():
"""Show SDK capabilities"""
print("🚀 Superior SDK Demo")
print("=" * 50)
async with AdvancedSDK("https://jsonplaceholder.typicode.com") as sdk:
print("n📥 Testing GET request with caching...")
response1 = await sdk.get("/posts/1")
print(f"First request - Standing: {response1.status_code}")
print(f"Title: {response1.knowledge.get('title', 'N/A')}")
response2 = await sdk.get("/posts/1")
print(f"Second request (cached) - Standing: {response2.status_code}")
print("n📤 Testing POST request...")
new_post = {
"title": "Superior SDK Tutorial",
"physique": "This SDK demonstrates fashionable Python patterns",
"userId": 1
}
post_response = await sdk.put up("/posts", knowledge=new_post)
print(f"POST Standing: {post_response.status_code}")
print(f"Created put up ID: {post_response.knowledge.get('id', 'N/A')}")
print("n⚡ Testing batch requests with charge limiting...")
duties = []
for i in vary(1, 6):
duties.append(sdk.get(f"/posts/{i}"))
outcomes = await asyncio.collect(*duties)
print(f"Batch accomplished: {len(outcomes)} requests")
for i, lead to enumerate(outcomes, 1):
print(f" Submit {i}: {outcome.knowledge.get('title', 'N/A')[:30]}...")
print("n❌ Testing error dealing with...")
attempt:
error_response = await sdk.get("/posts/999999")
print(f"Error response standing: {error_response.status_code}")
besides Exception as e:
print(f"Dealt with error: {kind(e).__name__}")
print("n✅ Demo accomplished efficiently!")
async def run_demo():
"""Colab-friendly demo runner"""
await demo_sdk()The demo_sdk coroutine walks by the SDK’s core options, issuing a cached GET request, performing a POST, executing a batch of GETs underneath charge limiting, and dealing with errors, towards the JSONPlaceholder API, printing standing codes and pattern knowledge as an example every functionality. The run_demo helper ensures this demo runs easily inside a Colab pocket book’s current occasion loop.
import nest_asyncio
nest_asyncio.apply()
if __name__ == "__main__":
attempt:
asyncio.run(demo_sdk())
besides RuntimeError:
loop = asyncio.get_event_loop()
loop.run_until_complete(demo_sdk())
class SDKBuilder:
"""Builder sample for SDK configuration"""
def __init__(self, base_url: str):
self.base_url = base_url
self.config = {}
def with_auth(self, api_key: str):
self.config['api_key'] = api_key
return self
def with_rate_limit(self, calls_per_minute: int):
self.config['rate_limit'] = calls_per_minute
return self
def construct(self) -> AdvancedSDK:
return AdvancedSDK(self.base_url, **self.config)Lastly, we apply nest_asyncio to allow nested occasion loops in Colab, then run the demo through asyncio.run (with a fallback to guide loop execution if wanted). It additionally introduces an SDKBuilder class that implements a fluent builder sample for simply configuring and instantiating the AdvancedSDK with customized authentication and rate-limit settings.
In conclusion, this SDK tutorial offers a scalable basis for any RESTful integration, combining fashionable Python idioms (dataclasses, async/await, context managers) with sensible tooling (charge limiter, cache, structured logging). By adapting the patterns proven right here, significantly the separation of considerations between request orchestration, caching, and response modeling, groups can speed up the event of latest API shoppers whereas making certain predictability, observability, and resilience.
Try the Codes. All credit score for this analysis goes to the researchers of this mission. Additionally, be at liberty to observe us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our Publication.
Sana Hassan, a consulting intern at Marktechpost and dual-degree pupil at IIT Madras, is enthusiastic about making use of expertise and AI to handle real-world challenges. With a eager curiosity in fixing sensible issues, he brings a contemporary perspective to the intersection of AI and real-life options.
