В этом руководстве мы покажем, как создать надёжный Python SDK, готовый к использованию в продакшене. Мы начнём с установки и настройки основных асинхронных HTTP-библиотек (aiohttp, nest-asyncio). Затем мы рассмотрим реализацию основных компонентов, включая структурированные объекты ответа, ограничение частоты запросов с помощью токенов, кэширование в памяти с TTL и чистый дизайн, основанный на dataclass.
Мы увидим, как объединить все эти части в класс AdvancedSDK, который поддерживает асинхронное управление контекстом, автоматическое повторение/ожидание при достижении лимита частоты запросов, внедрение заголовков JSON/auth и удобные методы HTTP-глаголов.
Установка и настройка асинхронной среды выполнения
Для начала импортируем asyncio и aiohttp, а также утилиты для работы со временем, обработкой JSON, моделированием dataclass, кэшированием (через hashlib и datetime) и структурированным логированием.
“`python
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
“`
Структурированный объект ответа
Класс APIResponse инкапсулирует детали HTTP-ответа, полезную нагрузку (data), код состояния, заголовки и метку времени получения в единый типизированный объект.
“`python
@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)
“`
Ограничение частоты запросов
Класс RateLimiter реализует простое ограничение частоты запросов с помощью политики токена-ведра.
“`python
class RateLimiter:
“””Token bucket rate limiter”””
def init(self, maxcalls: int = 100, timewindow: int = 60):
self.maxcalls = maxcalls
self.timewindow = timewindow
self.calls = []
def can_proceed(self) -> bool:
now = time.time()
self.calls = [calltime for calltime in self.calls if now – calltime < self.timewindow]
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]))
“`
Кэширование в памяти
Класс Cache предоставляет лёгкий механизм кэширования ответов API в памяти.
“`python
class Cache:
“””Simple in-memory cache with TTL”””
def init(self, default_ttl: int = 300):
self.cache = {}
self.defaultttl = defaultttl
def generatekey(self, method: str, url: str, params: Dict = None) -> str:
keydata = f”{method}:{url}:{json.dumps(params or {}, sortkeys=True)}”
return hashlib.md5(key_data.encode()).hexdigest()
def get(self, method: str, url: str, params: Dict = None) -> Optional[APIResponse]:
key = self.generatekey(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.generatekey(method, url, params)
expiry = datetime.now() + timedelta(seconds=ttl or self.default_ttl)
self.cache[key] = (response, expiry)
“`
AdvancedSDK
Класс AdvancedSDK объединяет всё вместе в чистый асинхронный клиент: он управляет сессией aiohttp через асинхронные контекстные менеджеры, внедряет заголовки JSON и auth, а также координирует RateLimiter и Cache.
“`python
class AdvancedSDK:
“””Advanced SDK with modern Python patterns”””
def init(self, baseurl: str, apikey: str = None, rate_limit: int = 100):
self.baseurl = baseurl.rstrip(‘/’)
self.apikey = apikey
self.session = None
self.ratelimiter = RateLimiter(maxcalls=rate_limit)
self.cache = Cache()
self.logger = self.setuplogger()
def setuplogger(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, exctype, excval, exc_tb):
“””Async context manager exit”””
if self.session:
await self.session.close()
def getheaders(self) -> Dict[str, str]:
headers = {‘Content-Type’: ‘application/json’}
if self.api_key:
headers[‘Authorization’] = f’Bearer {self.api_key}’
return headers
async def makerequest(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.ratelimiter.canproceed():
waittime = self.ratelimiter.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.getheaders()
) as resp:
responsedata = await resp.json() if resp.contenttype == ‘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.makerequest(‘GET’, endpoint, params=params, usecache=usecache)
async def post(self, endpoint: str, data: Dict = None) -> APIResponse:
return await self.makerequest(‘POST’, endpoint, data=data, use_cache=False)
async def put(self, endpoint: str, data: Dict = None) -> APIResponse:
return await self.makerequest(‘PUT’, endpoint, data=data, use_cache=False)
async def delete(self, endpoint: str) -> APIResponse:
return await self.makerequest(‘DELETE’, endpoint, use_cache=False)
“`
Это лишь некоторые из компонентов, которые мы рассмотрим в этом руководстве. Для более подробной информации и демонстрации возможностей SDK, пожалуйста, ознакомьтесь с полным кодом.
1. Какие основные асинхронные HTTP-библиотеки используются в данном руководстве для создания Python SDK?
В статье упоминаются библиотеки aiohttp и nest-asyncio.
2. Какие компоненты включает в себя класс APIResponse и для чего он предназначен?
Класс APIResponse инкапсулирует детали HTTP-ответа, полезную нагрузку (data), код состояния, заголовки и метку времени получения в единый типизированный объект.
3. Как работает механизм ограничения частоты запросов с помощью класса RateLimiter?
Класс RateLimiter реализует простое ограничение частоты запросов с помощью политики токена-ведра. В нём используется счётчик запросов и задаётся временной интервал, в течение которого количество запросов не должно превышать заданного максимума.
4. Какие методы используются для кэширования ответов API в памяти?
Для кэширования ответов API в памяти используется класс Cache, который предоставляет лёгкий механизм кэширования ответов API в памяти с TTL (время жизни).
5. Какие основные функции выполняет класс AdvancedSDK и как он объединяет все компоненты в чистый асинхронный клиент?
Класс AdvancedSDK объединяет все компоненты в чистый асинхронный клиент: он управляет сессией aiohttp через асинхронные контекстные менеджеры, внедряет заголовки JSON и auth, а также координирует RateLimiter и Cache.