Ozon Seller API в продакшене: что ломается и как чинить

В Tirebase мы выгружаем каталог шин (~250К SKU) на маркетплейсы, и Ozon --- один из основных каналов. Я ожидал потратить пару дней на интеграцию с Ozon Seller API. Документация есть, эндпоинты описаны. По факту провозился две недели, и часть проблем до сих пор решается костылями.

Проблема №1: Документация не соответствует реальным ответам

Документация Ozon API местами описывает поведение, которого нет. Поле visibility в ответе /v2/product/info задокументировано как обязательное, но для некоторых категорий возвращается null. Поле marketing_price иногда приходит как строка, иногда как число.

Я потерял два дня, отлаживая ошибку валидации Pydantic, прежде чем понял:

# Было — падало на реальных данных
class ProductInfo(BaseModel):
    marketing_price: float
    visibility: int

# Стало — работает с реальностью
class ProductInfo(BaseModel):
    marketing_price: float | str | None = None
    visibility: int | None = None

    @field_validator("marketing_price", mode="before")
    @classmethod
    def coerce_price(cls, v):
        if isinstance(v, str):
            return float(v) if v else None
        return v

Решение: я стал фиксировать каждое расхождение между документацией и реальным ответом. Со временем это превратилось в набор Pydantic-моделей, покрывающий все аномалии. Именно эти модели легли в основу python-ozon-api.

Проблема №2: Rate Limits без предупреждения

Ozon документирует лимиты в духе «не более 10 запросов в секунду». На практике это зависит от эндпоинта, времени суток и, кажется, фазы луны. Я получал 429 Too Many Requests при 3 запросах в секунду, а иногда 20 запросов проходили без проблем. Не уверен, что Ozon использует фиксированные лимиты --- похоже на какой-то динамический throttling.

Моё решение — адаптивный rate limiter с экспоненциальным backoff:

class AdaptiveRateLimiter:
    def __init__(self, initial_rps: float = 5.0, min_rps: float = 0.5):
        self.current_rps = initial_rps
        self.min_rps = min_rps
        self._semaphore = asyncio.Semaphore(int(initial_rps))
        self._window_start = time.monotonic()
        self._request_count = 0

    async def acquire(self):
        await self._semaphore.acquire()
        now = time.monotonic()
        elapsed = now - self._window_start
        if elapsed < 1.0:
            delay = 1.0 / self.current_rps
            await asyncio.sleep(delay)
        else:
            self._window_start = now
            self._request_count = 0

    def on_rate_limited(self):
        """Вызывается при получении 429"""
        self.current_rps = max(self.current_rps * 0.6, self.min_rps)
        logger.warning(f"Rate limited, снижаю RPS до {self.current_rps:.1f}")

    def on_success(self):
        """Постепенно восстанавливаем скорость"""
        self.current_rps = min(self.current_rps * 1.05, 10.0)

Ключевая идея --- при получении 429 мы не просто ждём, а снижаем общую скорость на 40%. А при успешных запросах плавно восстанавливаем её. Подход похож на TCP congestion control --- тот же AIMD (additive increase, multiplicative decrease). За месяц работы в продакшене количество ошибок 429 стало приемлемым --- единицы в сутки.

Проблема №3: Идемпотентность — враг мой

Некоторые эндпоинты Ozon идемпотентны, а некоторые — нет, и это нигде явно не указано. Повторный вызов /v1/product/import с тем же содержимым может создать дубликат товара, а может вернуть ошибку. Зависит от состояния задачи на стороне Ozon.

Я реализовал обёртку с отслеживанием задач:

class IdempotentTaskTracker:
    def __init__(self, redis: Redis):
        self.redis = redis

    async def execute_once(self, task_key: str, coro_factory):
        """Выполняет корутину только если задача ещё не запускалась"""
        lock_key = f"ozon:task:{task_key}"
        acquired = await self.redis.set(lock_key, "running", nx=True, ex=3600)
        if not acquired:
            existing = await self.redis.get(lock_key)
            if existing == "completed":
                logger.info(f"Задача {task_key} уже выполнена, пропускаю")
                return None
            # Задача запущена, но не завершена — ждём
            return await self._wait_for_completion(lock_key)

        try:
            result = await coro_factory()
            await self.redis.set(lock_key, "completed", ex=86400)
            return result
        except Exception:
            await self.redis.delete(lock_key)
            raise

Проблема №4: Изменение контракта без версионирования

В мае 2025 года Ozon изменил формат ответа /v2/product/list — поле items стало вложено в result.items вместо корневого уровня. Без предупреждения, без новой версии эндпоинта. Мой парсинг прайсов упал в 3 часа ночи.

После этого случая я реализовал защитный слой:

def extract_items(response: dict) -> list[dict]:
    """Извлекает items из ответа Ozon, независимо от вложенности"""
    if "result" in response and "items" in response["result"]:
        return response["result"]["items"]
    if "items" in response:
        return response["items"]
    # Последняя попытка — ищем первый список в ответе
    for value in response.values():
        if isinstance(value, list):
            logger.warning(f"Fallback extraction, key structure: {list(response.keys())}")
            return value
    return []

Грубо? Да. Но в продакшене грубое решение, которое работает в 3 часа ночи, лучше элегантного, которое падает.

Проблема №5: Кеширование ради выживания

Ozon API медленный. Получение информации о нескольких тысячах товаров занимает минуты из-за лимитов на размер батча (1000 товаров) и rate limits. При каждом обновлении цен мне нужна актуальная информация, но дёргать API каждый раз --- безумие.

Стратегия многоуровневого кеша (идея взята из cachetools + Redis):

class OzonCache:
    def __init__(self, redis: Redis):
        self.redis = redis
        self.local = TTLCache(maxsize=10000, ttl=60)  # L1: in-memory, 1 мин
        self.redis_ttl = 300  # L2: Redis, 5 мин

    async def get_product(self, product_id: int) -> ProductInfo | None:
        # L1 — моментальный доступ
        if product_id in self.local:
            return self.local[product_id]

        # L2 — Redis
        cached = await self.redis.get(f"ozon:product:{product_id}")
        if cached:
            product = ProductInfo.model_validate_json(cached)
            self.local[product_id] = product
            return product

        return None

    async def invalidate_by_category(self, category_id: int):
        """Инвалидация по категории — когда Ozon обновил правила"""
        pattern = f"ozon:product:cat:{category_id}:*"
        keys = []
        async for key in self.redis.scan_iter(pattern):
            keys.append(key)
        if keys:
            await self.redis.delete(*keys)

Рождение python-ozon-api

Все эти обходные пути, валидаторы, retry-стратегии и кеши постепенно сложились в библиотеку. Я вынес общий код, покрыл его тестами и опубликовал на PyPI. Честно говоря, не уверен, что архитектура библиотеки оптимальна --- я выделял код из продакшен-проекта, и некоторые абстракции выглядят натянуто.

Что вошло в библиотеку:

• Типизированные модели для основных эндпоинтов --- с учётом известных мне аномалий
• Встроенный rate limiter --- адаптивный, с экспоненциальным backoff
• Retry с jitter --- случайная задержка чтобы не создавать «стаю» повторных запросов (статья AWS хорошо объясняет зачем)
• Полностью async --- нативный aiohttp под капотом

Выводы

Работа с Ozon API научила меня важному принципу: никогда не доверяй чужому API. Всегда закладывай:

1. Валидацию на входе с мягкими типами
2. Retry с адаптивной скоростью
3. Кеширование для снижения зависимости
4. Мониторинг аномалий в ответах
5. Fallback-логику для изменённых контрактов

Ozon API --- далеко не худший (Wildberries API, например, ещё веселее). Но любой внешний API в продакшене --- это потенциальный источник проблем, и архитектура должна это учитывать с первого дня.