En este tutorial, guiamos a los usuarios a través de la construcción de un SDK de Python robusto y pronto para la producción. Comienza mostrando cómo instalar y configurar bibliotecas HTTP asíncronas esenciales (AIOHTTP, Nest-Asyncio). Luego camina a través de la implementación de componentes centrales, incluidos los objetos de respuesta estructurados, la acotación de la tasa de token-bucket, el almacenamiento en distinción en memoria con TTL y un diseño expedito y basado en dataclass. Veremos cómo concluir estas piezas en una clase AdvancedSDK que admite la papeleo del contexto Async, el comportamiento forzoso de reintento/esperanza de acotación, inyección de encabezados JSON/Auth y métodos convenientes HTTP-Verb. En el camino, un arnés de demostración contra JsonplaceHolder ilustra la eficiencia del almacenamiento en distinción, la producción de lotes con límites de velocidad, el manejo de errores e incluso muestra cómo extender el SDK a través de un patrón fluido de «constructor» para la configuración personalizada.
import asyncio
import aiohttp
import time
import json
from typing import Dict, List, Optional, Any, Union
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
import hashlib
import logging
!pip install aiohttp nest-asyncioInstalamos y configuramos el tiempo de ejecución asíncrono importando Asyncio y AIOHTTP, pegado con utilidades para el tiempo, manejo JSON, modelado de dataclass, almacenamiento en distinción (a través de Hashlib y DateTime) y registro estructurado. La hilera! Pip install Aiohttp Nest-Asyncio asegura que el cuaderno pueda ejecutar un tirabuzón de evento sin problemas interiormente de Colab, lo que permite solicitudes robustas de async HTTP y flujos de trabajo limitados.
@dataclass
class APIResponse:
"""Structured response object"""
data: Any
status_code: int
headers: Dict(str, str)
timestamp: datetime
def to_dict(self) -> Dict:
return asdict(self)El DatacLass de APIRESPOND encapsula los detalles de respuesta HTTP, la carga útil (datos), el código de estado, los encabezados y la marca de tiempo de recuperación en un solo objeto mecanografiado. El ayudante To_Dict () convierte la instancia en un diccionario simple para un acomodaticio registro, serialización o procesamiento posterior.
class RateLimiter:
"""Token bucket rate 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)))La clase RatElimiter impone una política simple de token-bucket al rastrear las marcas de tiempo de las llamadas recientes y permitir que Max_Calls interiormente de un Time_Window rodante. Cuando se alcanza el frontera, can_proced () devuelve hipócrita y Wait_time () calcula cuánto tiempo se detiene antiguamente de hacer la próximo solicitud.
class Cache:
"""Simple in-memory cache with TTL"""
def __init__(self, default_ttl: int = 300):
self.cache = {}
self.default_ttl = default_ttl
def _generate_key(self, method: str, url: str, params: Dict = None) -> str:
key_data = f"{method}:{url}:{json.dumps(params or {}, sort_keys=True)}"
return hashlib.md5(key_data.encode()).hexdigest()
def get(self, method: str, url: str, params: Dict = None) -> Optional(APIResponse):
key = self._generate_key(method, 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, method: str, url: str, response: APIResponse, params: Dict = None, ttl: int = None):
key = self._generate_key(method, url, params)
expiry = datetime.now() + timedelta(seconds=ttl or self.default_ttl)
self.cache(key) = (response, expiry)La clase de distinción proporciona un distinción TTL en memoria informal para las respuestas de API mediante el hash de la firma de solicitud (Método, URL, Params) en una secreto única. Devuelve los objetos de apagoses en distinción válidos antiguamente de la expiración y desaloja automáticamente las entradas obsoletas posteriormente de que haya transcurrido su tiempo de vida.
class AdvancedSDK:
"""Advanced SDK with modern 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 - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
return logger
async def __aenter__(self):
"""Async context manager entry"""
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Async context manager exit"""
if self.session:
await self.session.close()
def _get_headers(self) -> Dict(str, str):
headers = {'Content-Type': 'application/json'}
if self.api_key:
headers('Authorization') = f'Bearer {self.api_key}'
return headers
async def _make_request(self, method: str, endpoint: str, params: Dict = None,
data: Dict = None, use_cache: bool = True) -> APIResponse:
"""Core request method with rate limiting and caching"""
if use_cache and method.upper() == 'GET':
cached = self.cache.get(method, endpoint, params)
if cached:
self.logger.info(f"Cache hit for {method} {endpoint}")
return cached
if not self.rate_limiter.can_proceed():
wait_time = self.rate_limiter.wait_time()
self.logger.warning(f"Rate limit hit, waiting {wait_time:.2f}s")
await asyncio.sleep(wait_time)
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try:
async with self.session.request(
method=method.upper(),
url=url,
params=params,
json=data,
headers=self._get_headers()
) as resp:
response_data = await resp.json() if resp.content_type == 'application/json' else await resp.text()
api_response = APIResponse(
data=response_data,
status_code=resp.status,
headers=dict(resp.headers),
timestamp=datetime.now()
)
if use_cache and method.upper() == 'GET' and 200 <= resp.status < 300:
self.cache.set(method, endpoint, api_response, params)
self.logger.info(f"{method.upper()} {endpoint} - Status: {resp.status}")
return api_response
except Exception as e:
self.logger.error(f"Request failed: {str(e)}")
raise
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 post(self, endpoint: str, data: Dict = None) -> APIResponse:
return await self._make_request('POST', endpoint, data=data, use_cache=False)
async def put(self, endpoint: str, data: Dict = None) -> APIResponse:
return await self._make_request('PUT', endpoint, data=data, use_cache=False)
async def delete(self, endpoint: str) -> APIResponse:
return await self._make_request('DELETE', endpoint, use_cache=False)La clase AdvancedSDK envuelve todo en un cliente expedito y async-primero: administra una sesión AIOHTTP a través de los gerentes de contexto Async, inyecta los encabezados JSON y Auth, y coordina nuestro ratelimiter y distinción debajo del capó. Su método _make_request centraliza Get/Post/Put/Eliminar la método, el manejo de las búsquedas de distinción, las esperas de frontera de velocidad, el registro de errores y el empaque de respuesta en objetos APIRSponse, mientras que los ayudantes Get/Post/Put/Eliminar nos dan llamadas ergonómicas y de suspensión nivel.
async def demo_sdk():
"""Demonstrate SDK capabilities"""
print("🚀 Advanced 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 - Status: {response1.status_code}")
print(f"Title: {response1.data.get('title', 'N/A')}")
response2 = await sdk.get("/posts/1")
print(f"Second request (cached) - Status: {response2.status_code}")
print("n📤 Testing POST request...")
new_post = {
"title": "Advanced SDK Tutorial",
"body": "This SDK demonstrates modern Python patterns",
"userId": 1
}
post_response = await sdk.post("/posts", data=new_post)
print(f"POST Status: {post_response.status_code}")
print(f"Created post ID: {post_response.data.get('id', 'N/A')}")
print("n⚡ Testing batch requests with rate limiting...")
tasks = ()
for i in range(1, 6):
tasks.append(sdk.get(f"/posts/{i}"))
results = await asyncio.gather(*tasks)
print(f"Batch completed: {len(results)} requests")
for i, result in enumerate(results, 1):
print(f" Post {i}: {result.data.get('title', 'N/A')(:30)}...")
print("n❌ Testing error handling...")
try:
error_response = await sdk.get("/posts/999999")
print(f"Error response status: {error_response.status_code}")
except Exception as e:
print(f"Handled error: {type(e).__name__}")
print("n✅ Demo completed successfully!")
async def run_demo():
"""Colab-friendly demo runner"""
await demo_sdk()El Coroutine Demo_SDK camina a través de las características centrales del SDK, emitiendo una solicitud GET en distinción, realizar una publicación, ejecutar un parte de parte de acotación de tasas y manejo de errores, contra la API JSONplaceHolder, los códigos de estado de impresión y los datos de muestra para ilustrar cada capacidad. Run_demo Helper asegura que esta demostración se ejecute sin problemas interiormente del tirabuzón de eventos existente de un cuaderno de Colab.
import nest_asyncio
nest_asyncio.apply()
if __name__ == "__main__":
try:
asyncio.run(demo_sdk())
except RuntimeError:
loop = asyncio.get_event_loop()
loop.run_until_complete(demo_sdk())
class SDKBuilder:
"""Builder pattern 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 build(self) -> AdvancedSDK:
return AdvancedSDK(self.base_url, **self.config)Finalmente, aplicamos Nest_asyncio para habilitar los bucles de eventos anidados en Colab, luego ejecutar la demostración a través de Asyncio.run (con una ejecución de tirabuzón de retroceso a manual si es necesario). Asimismo presenta una clase SDKBuilder que implementa un patrón de constructor con fluidez para configurar e instanciar fácilmente el AdvancedSDK con la autenticación personalizada y la configuración de frontera de velocidad.
En conclusión, este tutorial SDK proporciona una pulvínulo escalable para cualquier integración tranquila, que combina modernos idiomas de Python (dataclasses, async/esperanza, gerentes de contexto) con herramientas prácticas (limitador de velocidad, distinción, registro estructurado). Al adaptar los patrones que se muestran aquí, particularmente la separación de las preocupaciones entre la orquestación de solicitudes, el almacenamiento en distinción y el modelado de respuesta, los equipos pueden acelerar el ampliación de nuevos clientes de API al tiempo que garantizan previsibilidad, observabilidad y resistor.
Mira el Codos. Todo el crédito por esta investigación va a los investigadores de este plan. Adicionalmente, siéntete osado de seguirnos Gorjeo Y no olvides unirte a nuestro Subreddit de 100k+ ml y suscribirse a Nuestro boletín.
Sana Hassan, una pasante de consultoría en MarktechPost y estudiante de doble calidad en IIT Madras, le apasiona aplicar tecnología e IA para invadir los desafíos del mundo vivo. Con un gran interés en resolver problemas prácticos, aporta una nueva perspectiva a la intersección de la IA y las soluciones de la vida vivo.
