Создание производительного асинхронного Python SDK с ограничением скорости и кэшированием

Руководство по созданию готового к производству асинхронного Python SDK с ограничением скорости, кэшированием в памяти и аутентификацией

Создание надежного и производительного SDK для работы с API — задача, требующая внимания к деталям и понимания современных технологий. В этом руководстве мы рассмотрим, как разработать асинхронный Python SDK, который будет включать в себя такие важные компоненты, как ограничение скорости, кэширование в памяти и аутентификация. Мы будем использовать библиотеки aiohttp и nest-asyncio, чтобы обеспечить эффективное выполнение HTTP-запросов.

Установка и настройка

Первым шагом будет установка необходимых библиотек. Выполните следующую команду:

!pip install aiohttp nest-asyncio

Эта команда установит асинхронные библиотеки, которые позволят нам работать с HTTP-запросами и управлять потоками событий.

Реализация основных компонентов

Структурированный объект ответа

Для начала создадим класс APIResponse, который будет хранить детали ответа от API:

from dataclasses import dataclass
from typing import Any, Dict
from datetime import datetime

@dataclass
class APIResponse:
    data: Any
    status_code: int
    headers: Dict[str, str]
    timestamp: datetime

    def to_dict(self) -> Dict:
        return asdict(self)

Этот класс инкапсулирует информацию о HTTP-ответе, включая полезную нагрузку, код состояния и заголовки.

Ограничение скорости

Следующий шаг — реализация механизма ограничения скорости. Мы создадим класс RateLimiter, который будет управлять количеством запросов:

import time

class RateLimiter:
    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]))

Класс RateLimiter реализует политику токенов, позволяя контролировать количество запросов к API.

Кэширование в памяти

Теперь добавим кэширование для повышения производительности. Создадим класс Cache, который будет хранить ответы:

from datetime import timedelta
import hashlib
import json

class Cache:
    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)

Класс Cache предоставляет легковесное кэширование с ограничением по времени жизни (TTL) для ответов API.

Расширенный класс SDK

Теперь объединим все компоненты в класс AdvancedSDK, который будет управлять сессиями и аутентификацией:

import aiohttp
import logging

class AdvancedSDK:
    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):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

Класс AdvancedSDK интегрирует все компоненты, управляя сессиями и заголовками, а также координируя ограничение скорости и кэширование.

Демонстрация возможностей SDK

Теперь давайте посмотрим, как использовать наш SDK:

async def demo_sdk():
    async with AdvancedSDK("https://jsonplaceholder.typicode.com") as sdk:
        response = await sdk.get("/posts/1")
        print(f"Status: {response.status_code}, Title: {response.data.get('title', 'N/A')}")

Функция demo_sdk демонстрирует основные функции SDK, выполняя GET-запросы и показывая кэширование и обработку ошибок.

Заключение

Это руководство по созданию SDK предоставляет масштабируемую основу для любой интеграции RESTful, сочетая современные практики программирования на Python с практическими инструментами. Применяя описанные паттерны, команды разработчиков могут ускорить создание новых клиентов API, обеспечивая предсказуемость, наблюдаемость и устойчивость.

Для получения дополнительных ресурсов и обсуждений в сообществе, пожалуйста, следите за нами в Twitter и присоединяйтесь к нашему сабреддиту по машинному обучению.