This tutorial will information you to construct a strong, production-able Python SDK. Let’s begin by exhibiting you set up and configure the necessary asynchronous HTTP library (AIOHTTP, nest-asyncio). Subsequent, we proceed with implementing core parts, together with structured response objects, token bucket charge limits, in-memory cache utilizing TTL, and clear datalas-driven designs. Discover ways to wrap these items in an AdvancedSDK class that helps asynchronous context administration, automated retry/wait/auth header injection, and handy HTTP-Verb strategies. Alongside the best way, the demo harness for JSONPlaceHolder exhibits lengthen the SDK through Fluent’s “Builder” sample for cache effectivity, rate-limited batch fetch, error dealing with, and even customized configurations.
import asyncio
import aiohttp
import time
import json
from typing import Dict, Listing, Non-compulsory, Any, Union
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
import hashlib
import logging
!pip set up aiohttp nest-asyncioImport Asyncio and AIOHTTP to put in and configure the asynchronous runtime, together with utilities for timing, JSON dealing with, Dataclass modeling, caching (through Hashlib and DateTime), and structured logging. ! The pip set up aiohttp nest-asyncio line permits the pocket book to seamlessly execute occasion loops throughout the colab, permitting for sturdy non-yync http requests and rate-limiting 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)Apiresponse Dataclass encapsulates HTTP response particulars, payload (knowledge), standing code, header, and retrieval timestamps into an object entered as a single kind. The to_dict() helper converts situations right into a plain dictionary for easy 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 current calls and permitting it to be max_calls throughout the Rolling Time_Window. When the restrict is reached, can_proceed() returns false, and wait_time() calculates the time 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, methodology: str, url: str, params: Dict = None) -> str:
key_data = f"{methodology}:{url}:{json.dumps(params or {}, sort_keys=True)}"
return hashlib.md5(key_data.encode()).hexdigest()
def get(self, methodology: str, url: str, params: Dict = None) -> Non-compulsory[APIResponse]:
key = self._generate_key(methodology, 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, methodology: str, url: str, response: APIResponse, params: Dict = None, ttl: int = None):
key = self._generate_key(methodology, 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 request signatures (Methodology, URL, Params) to a novel key. Returns a legitimate cached Apiresponse object earlier than it expires, robotically ejecting outdated entries after a time period has handed.
class AdvancedSDK:
"""Superior SDK with trendy 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-Sort': 'software/json'}
if self.api_key:
headers['Authorization'] = f'Bearer {self.api_key}'
return headers
async def _make_request(self, methodology: str, endpoint: str, params: Dict = None,
knowledge: Dict = None, use_cache: bool = True) -> APIResponse:
"""Core request methodology with charge limiting and caching"""
if use_cache and methodology.higher() == 'GET':
cached = self.cache.get(methodology, endpoint, params)
if cached:
self.logger.data(f"Cache hit for {methodology} {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('/')}"
strive:
async with self.session.request(
methodology=methodology.higher(),
url=url,
params=params,
json=knowledge,
headers=self._get_headers()
) as resp:
response_data = await resp.json() if resp.content_type == 'software/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 methodology.higher() == 'GET' and 200 <= resp.standing < 300:
self.cache.set(methodology, endpoint, api_response, params)
self.logger.data(f"{methodology.higher()} {endpoint} - Standing: {resp.standing}")
return api_response
besides Exception as e:
self.logger.error(f"Request failed: {str(e)}")
elevate
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 brings every part collectively right into a clear, asynchronous consumer. Handle AIOHTTP classes through asynchronous supervisor, inject JSON and AUTH headers, and alter the Ratelimiter and cache below Hood. Its _make_request methodology focuses on the apiresponse object for Get/put up/put/deletion logic, cache lookup, charge restrict wait, error logging, and response packing.
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 trendy 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, end in enumerate(outcomes, 1):
print(f" Put up {i}: {outcome.knowledge.get('title', 'N/A')[:30]}...")
print("n❌ Testing error dealing with...")
strive:
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()DEMO_SDK Coroutine goes via the SDK core performance, points cached GET requests, carry out posts, limits GetS limits, handles errors, handles errors towards the JSONPlaceHolder API, pattern knowledge to clarify every skill. The run_demo helper ensures that this demo runs easily throughout the current occasion loop of the Colob Notice.
import nest_asyncio
nest_asyncio.apply()
if __name__ == "__main__":
strive:
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, apply nest_asyncio to allow nested occasion loops in colab earlier than operating the demo through asyncio.run (utilizing handbook loop execution if essential). It additionally introduces the SDKBuilder class that implements a fluent builder sample to simply configure and instantiate AdvancedSDK utilizing customized authentication and charge limiting settings.
In conclusion, this SDK tutorial offers a scalable basis for any consolation integration, combining the newest Python Idioms (Dataclasses, Async/await, Context Managers) with sensible instruments (charge limiting, caching, structural logging). The patterns proven right here, significantly the separation of considerations between request orchestration, caching, and response modeling, enable groups to make sure predictability, observability and resilience whereas accelerating the event of latest API purchasers.
Please examine code. All credit for this examine will likely be directed to researchers on this undertaking. Additionally, please be at liberty to observe us Twitter And do not forget to affix us 100k+ ml subreddit And subscribe Our Newsletter.
Sana Hassan, a consulting intern at MarkTechPost and a dual-level pupil at IIT Madras, is obsessed with making use of know-how and AI to handle real-world challenges. With a robust curiosity in fixing actual issues, he brings a brand new perspective to the intersection of AI and actual options.
