On this tutorial, we information customers by means of constructing a strong, production-ready Python SDK. It begins by displaying the best way to set up and configure important asynchronous HTTP libraries (aiohttp, nest-asyncio). It then walks by means of the implementation of core parts, together with structured response objects, token-bucket price limiting, in-memory caching with TTL, and a clear, dataclass-driven design. We’ll see the best way 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 way in which, a demo harness in opposition to JSONPlaceholder illustrates caching effectivity, batch fetching with price limits, error dealing with, and even reveals the best way to lengthen the SDK by way of a fluent “builder” sample for customized configuration.
import asyncio
import aiohttp
import time
import json
from typing import Dict, Record, Non-obligatory, 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 (by way of 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 strong async HTTP requests and rate-limited workflows.
@dataclass
class APIResponse:
"""Structured response object"""
information: 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 (information), 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 straightforward logging, serialization, or downstream processing.
class RateLimiter:
"""Token bucket price 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 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 current 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) -> Non-obligatory[APIResponse]:
key = self._generate_key(technique, url, params)
if key in self.cache:
response, expiry = self.cache[key]
if datetime.now() The Cache class gives a light-weight in-memory TTL cache for API responses by hashing the request signature (technique, URL, params) into a novel key. It returns legitimate cached APIResponse objects earlier than expiry and routinely 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,
information: Dict = None, use_cache: bool = True) -> APIResponse:
"""Core request technique with price limiting and caching"""
if use_cache and technique.higher() == 'GET':
cached = self.cache.get(technique, endpoint, params)
if cached:
self.logger.information(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=information,
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(
information=response_data,
status_code=resp.standing,
headers=dict(resp.headers),
timestamp=datetime.now()
)
if use_cache and technique.higher() == 'GET' and 200 APIResponse:
return await self._make_request('GET', endpoint, params=params, use_cache=use_cache)
async def submit(self, endpoint: str, information: Dict = None) -> APIResponse:
return await self._make_request('POST', endpoint, information=information, use_cache=False)
async def put(self, endpoint: str, information: Dict = None) -> APIResponse:
return await self._make_request('PUT', endpoint, information=information, 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 shopper: it manages an aiohttp session by way of async context managers, injects JSON and auth headers, and coordinates our RateLimiter and Cache beneath 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/submit/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.information.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.submit("/posts", information=new_post)
print(f"POST Standing: {post_response.status_code}")
print(f"Created submit ID: {post_response.information.get('id', 'N/A')}")
print("n⚡ Testing batch requests with price 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, end in enumerate(outcomes, 1):
print(f" Put up {i}: {outcome.information.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 means of the SDK’s core options, issuing a cached GET request, performing a POST, executing a batch of GETs beneath price limiting, and dealing with errors, in opposition to the JSONPlaceholder API, printing standing codes and pattern information 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 by way of 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 gives a scalable basis for any RESTful integration, combining fashionable Python idioms (dataclasses, async/await, context managers) with sensible tooling (price 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 guaranteeing predictability, observability, and resilience.
Try the Codes. All credit score for this analysis goes to the researchers of this mission. Additionally, be happy 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 know-how and AI to handle real-world challenges. With a eager curiosity in fixing sensible issues, he brings a recent perspective to the intersection of AI and real-life options.
