Руководство по созданию асинхронного Python SDK, готового к использованию в продакшене, с ограничением частоты запросов, кэшированием в

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

В этом руководстве мы покажем, как создать надёжный 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 window]

if len(self.calls) 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() generatekey(method, url, params)

...

Читать далее