OzonAPI

🌀 OzonAPI

Предоставляет легковесный асинхронный pydantic-интерфейс для взаимодействия с Ozon Seller API.

Поддерживает ограничения запросов, работу с несколькими кабинетами, классическую и OAuth-авторизацию и гибкую настройку через .env.

✅ Актуально на 3-й квартал 2026 года.

🤝 Коммиты приветствуются!

⚡️ Применение

• 📊 Автоматизация сквозной аналитики и отчетов: Excel, Google Sheets, Bitrix24, Power BI
• 🔄 Промежуточное программное обеспечение: 1С, ERP, WMS, CRM и брокеры
• 🤖 Чат-боты: Telegram, Max (потенциально), RAG (для актуализации контекста)
• 📦 PIM-системы: управлениe контентом, товарами, ценами, остатками и отзывами
• 🗄️ DataLake: товары, цены, остатки, заказы и доставки

🚀 Основные возможности и характеристики

• ✔️ Проверено в production - используется на боевых кабинетах продавцов Ozon
• 🔑 Универсальная авторизация - поддерживает как client_id и api_key, так и OAuth-токен
• ⚡️ Асинхронный дизайн - построен на aiohttp для высокопроизводительных операций
• 👍 Простое использование - быстрое развертывание и интеграция с вашим проектом
• 📝 Отличная документация - все методы содержат подробное описание и примеры
• 👥 Мульти-аккаунт - одновременная работа с несколькими кабинетами продавца Ozon
• ⏱️ Умное ограничение запросов - автоматическое соблюдение лимитов API Ozon
• ➿ Гибкая настройка лимитов - кастомные лимиты запросов для методов
• ⏳ Кеширование ответов - для методов, возвращающих статичные данные
• 🔄 Автоповторы - экспоненциальные повторные попытки запросов при сбоях
• 🛡️ Валидация данных - строгая типизация с Pydantic v2
• 📊 Детальное логирование - асинхронная трассировка операций
• 🎪 Гибкая конфигурация - настройка через классы, переменные окружения или .env файл
• 🧹 Очистка ресурсов - автоматический контроль посредством контекстных менеджеров
• 🧪 Полное покрытие тестами - вся основная функциональность

🤝 Данный проект является форком с глубокой доработкой и актуализацией проекта python-ozon-api от mephistofox:

⚙️ Быстрый старт

Установка

pip install ozonapi-async

Базовое использование

import asyncio
from ozonapi import SellerAPI

async def main():
    async with SellerAPI(
        client_id="your_client_id",     # Обязательно, если не указан oauth-токен
        api_key="your_api_key",         # Обязательно, если не указан oauth-токен
        # token="your_oauth_token",     # Альтернативная авторизация с oauth-токеном
    ) as api:
        # Получение информации о владельце личного кабинета
        seller_info = await api.seller_info()
        
        print(f"Компания: {seller_info.company.name}")
        print(f"ИНН: {seller_info.company.inn}")

if __name__ == "__main__":
    asyncio.run(main())

Настройка через конфигурационный класс

- client_id: Идентификатор клиента Ozon (опционально, не требуется при указании token)
- api_key: Авторизационный ключ Ozon Seller API (опционально, не требуется при указании token)
- token: OAuth-токен Ozon Seller API (опционально, не требуется при указании api_key)
- base_url: Базовый URL API Ozon (опционально)
- max_requests_per_second: Максимальное количество запросов в секунду (опционально, по умолчанию 27)
- min_instance_ttl: Длительность памяти об активных клиентах в секундах для ограничения запросов (опционально)
- connector_limit: Лимит одновременных соединений для клиента (опционально, по умолчанию 100)
- request_timeout: Максимальное время ожидания ответа на запрос в секундах (опционально, по умолчанию 30)
- max_retries: Максимальное количество повторных попыток для неудачных запросов (опционально, по умолчанию 5)
- retry_min_wait: Минимальная задержка между повторами неудачных запросов в секундах (опционально, по умолчанию 2)
- retry_max_wait: Максимальная задержка между повторами неудачных запросов в секундах (опционально, по умолчанию 10)
- log_level: Уровень логирования (опционально, по умолчанию ERROR)
- log_json: Выводить в JSON (опционально)
- log_format: Формат лога (опционально)
- log_use_async: True, чтобы включить асинхронный режим (опционально, по умолчанию True)
- log_max_queue_size: Максимальный размер очереди (опционально, только для асинхронного режима)
- log_dir: Путь к директории с логами (опционально)
- log_file: Имя файла логов (опционально, при указании логирует в файл)
- log_max_bytes: Максимальный размер файла логов в байтах (опционально, по умолчанию 10M)
- log_backup_files_count: Кол-во файлов архивных логов, которые нужно хранить (опционально, по умолчанию 5)

import asyncio
from ozonapi import SellerAPI, SellerAPIConfig

async def main():
    # Создание кастомной конфигурации
    config = SellerAPIConfig(
        token="your_oauth_token",
        log_level="DEBUG"
    )
    
    async with SellerAPI(config=config) as api:
        # Работа с товарами
        products = await api.product_list()
        # Работа с ценами
        prices = await api.product_info_prices()

asyncio.run(main())

Настройка через .env файл

• Любой из параметров конфигурации можно задать в файле .env, расположенном в корне вашего проекта.
• Правило наименования параметров в .env: префикс OZON_SELLER_ + имя параметра в верхнем регистре.

Например, для client_id строка в файле .env примет вид: OZON_SELLER_CLIENT_ID=1234556

Создайте файл .env в корне проекта:

OZON_SELLER_CLIENT_ID=your_client_id
OZON_SELLER_API_KEY=your_api_key
OZON_SELLER_MAX_REQUESTS_PER_SECOND=30
OZON_SELLER_REQUEST_TIMEOUT=60.0
OZON_SELLER_MAX_RETRIES=5

Использование с автоматической загрузкой из .env:

import asyncio
from ozonapi import SellerAPI

async def main():
    # Конфигурация автоматически загружается из .env
    async with SellerAPI() as api:
        # Ваши API вызовы
        warehouses = await api.warehouse_list()
        print(f"Доступно складов: {len(warehouses.warehouses)}")

asyncio.run(main())

Пример использования

"""
Задача: Выгрузить описания для всех товаров продавца.

Описание решения: 
Частями выгрузим список товаров методом product_list(), 
чтобы узнать их product_id, и по мере выгрузки 
асинхронно получим описания для товаров с помощью метода 
product_info_description().

Для этих целей используем очереди из модуля asyncio и обработаем
данные по схеме producer -> queue -> consumers.

Сделаем одного производителя, чтобы складывал идентификаторы в очередь,
и несколько потребителей, каждый из которых будет забирать по одному 
из очереди и асинхронно выгружать его описание.
"""

import asyncio
from pprint import pprint

from ozonapi import SellerAPI, SellerAPIConfig
from ozonapi.seller.schemas.products import ProductListRequest, ProductListResponse,

ProductInfoDescriptionRequest

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

product_list_limit = 10     # Кол-во товаров, выгружаемых за одну итерацию
consumers_amount = 5        # Кол-во потребителей, выгружающих описания
consumers_rps_max_limit = 6 # Максимальное кол-во запросов в секунду для каждого потребителя
queue_max_size = product_list_limit * consumers_amount  # Максимальный размер очереди

product_descriptions = list()


async def producer(queue):
    """Получает батчи из списка товаров и добавляет product_id в очередь на получение описания."""

    async with SellerAPI(
            config=SellerAPIConfig(
                # Понижаем уровень логирования (для наглядности)
                log_level="INFO"
            )
    ) as api:

        # Параметры, необходимые для выборки данных
        products_count = 0  # Счетчик выбранных товаров
        last_id = str()  # Идентификатор для пагинации

        while True:
            # Отправляем запрос и получаем очередную партию данных о товарах
            products_batch: ProductListResponse = await api.product_list(
                ProductListRequest(
                    limit=product_list_limit,
                    last_id=last_id
                ),
            )

            # Переопределяем идентификатор выборки для следующей итерации
            last_id = products_batch.result.last_id
            # Увеличиваем счетчик выбранных товаров
            products_count += len(products_batch.result.items)

            # Добавляем айдишник каждого товара в очередь на выгрузку описания
            for item in products_batch.result.items:
                await queue.put(item.product_id)

            # Отчитываемся в консоль
            api.logger.info(
                f"Добавлено в обработку {products_count} из {products_batch.result.total} элементов."
            )

            # Если выгрузили все товары, то прерываем цикл
            if products_count == products_batch.result.total:
                break


async def consumer(queue):
    """Получает айдишники товаров из очереди и получает описания."""

    async with SellerAPI(
            config=SellerAPIConfig(
                # Понижаем уровень логирования
                log_level="INFO",
                # Ограничиваем кол-во запросов
                max_requests_per_second=consumers_rps_max_limit
            )
    ) as api:
        while True:
            # Достаем идентификатор очередного товара из очереди
            product_id = await queue.get()

            # Отправляем запрос и получаем ответ с описанием
            product_description = await api.product_info_description(
                ProductInfoDescriptionRequest(
                    product_id=product_id
                ),
            )

            # Добавляем полученное описание в общий список 
            # и выводим в консоль отчет о выполнении
            product_descriptions.append(product_description.result)
            api.logger.info(f"Получено описание для товара {product_description.result.id}")

            # Обновляем счетчик задач
            queue.task_done()


async def main() -> None:
    # Создаем очередь заданий на выгрузку описаний
    queue = asyncio.Queue(maxsize=queue_max_size)

    # Формируем заданное кол-во потребителей (выгружают описания) 
    [asyncio.create_task(consumer(queue)) for _ in range(consumers_amount)]

    # Запускаем производителя (узнает product_id товаров и заказывает для них описания)
    await producer(queue)

    # Дожидаемся, когда потребители обработают все задачи в очереди
    await queue.join()


if __name__ == '__main__':
    asyncio.run(main())

    # Выводим полученные описания
    for product_description in product_descriptions:
        pprint(product_description.model_dump())

Работа с несколькими аккаунтами

import asyncio
from ozonapi import SellerAPI

async def main():
    # Создание клиентов для разных аккаунтов
    configs = [
        {"client_id": "id1", "api_key": "key1"},
        {"client_id": "id2", "api_key": "key2"}
    ]
    
    tasks = []
    for config in configs:
        task = asyncio.create_task(fetch_account_data(config))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)

async def fetch_account_data(config):
    async with SellerAPI(**config) as api:
        products = await api.product_list()
        return {
            "client_id": config["client_id"],
            "product_count": len(products.result.items)
        }

asyncio.run(main())

🏗️ Архитектурные особенности

Управление ограничениями запросов

Проект автоматически соблюдает лимиты API Ozon с возможностью тонкой настройки:

# Кастомные лимиты
config = SellerAPIConfig(
    max_requests_per_second=25,  # Безопасный запас
    retry_min_wait=2.0,
    retry_max_wait=10.0
)

💡 Обратите внимание:

• Реализовано 2 вида ограничителей запросов: общий для всех запросов с одним client_id или token, а также для каждого экземпляра SellerAPI.
• Общий ограничитель запросов определен в SellerAPIConfig.max_requests_per_second и может быть переопределен в .env OZON_SELLER_MAX_REQUESTS_PER_SECOND.
• Помимо общего ограничителя можно задавать ограничители запросов для каждого инстанса SellerAPI с помощью max_requests_per_second.
• Если кол-во запросов по всем инстансам одного client_id или token превысит значение общего ограничителя, то будет применен общий ограничитель.
• 50 запросов в сек. — определенное документацией суммарное ограничение на все выполняемые запросы от всех методов с одного client_id в единицу времени.
• 25-27 запросов в сек. - оптимальное значение для ненагруженных API-запросами кабинетов (получено экпериментальным путем).
• У многих методов есть свои ограничения, которые не описаны документацией и могут динамически меняться Ozon, в зависимости от нагрузки на сервера.

Обработка ошибок и повторные попытки

Автоматические повторы запросов с экспоненциальной задержкой:

# Настройка стратегии повторов
config = SellerAPIConfig(
    max_retries=3,           # Максимум 3 попытки
    retry_min_wait=1.0,      # Минимальная задержка
    retry_max_wait=10.0      # Максимальная задержка
)

💡 Обратите внимание: Логер уведомляет об ошибке (например, при превышении кол-ва запросов), при этом планируется повторная отправка запроса, который инициировал исключение.

⚠️ Важные примечания

Производительность

Текущая реализация оптимизирована для однопоточного асинхронного использования. Для мультипроцессных сценариев требуется дополнительная настройка системы ограничения запросов.

Лимиты API

Проект автоматически соблюдает официальные лимиты Ozon API, но рекомендуется использовать консервативные настройки лимитов (оптимально 25-27 запросов в сек. в сумме)

🔧 Разработка

Установка

git clone https://github.com/a-ulianov/OzonAPI.git
cd OzonAPI
pip install -e
pytest --cov=ozonapi --cov-report=html

🤝 Поддержка

• Документация: GitHub Repository
• Issues: GitHub Issues
• Changelog: Releases
• Telegram: OzonAPI Async

✔️ Реализованные методы Ozon Seller API

Реализовано методов: 463 / 463

README 内容过长，已截取前 64 KB。完整内容请前往 GitHub 仓库查看。