ozapi

Внутренний типизированный Python-клиент для официальных Seller API и Performance API Ozon.

Сейчас библиотека поддерживает получение токена и статистики Performance API, а также получение списка и подробной информации о товарах, остатков FBO по складам, данных аналитики, начислений за день, создание отчёта о стоимости размещения по товарам и получение информации о созданном отчёте Seller API. Реализация сверяется с сохранёнными спецификациями:

• specs/upstream/seller_swagger.json;
• specs/upstream/performance_swagger.json.

Ответы методов возвращаются без преобразования в виде обычных dict и list. TypedDict используется только для подсказок IDE и статической проверки; неизвестные поля ответа сохраняются.

Установка

Пакет требует Python 3.14 или новее:

pip install ozapi

При использовании uv:

uv add ozapi

Установка для разработки

uv sync

Требуется Python 3.14 или новее.

Структура API

Seller API и Performance API разделены на независимые семейства ресурсов и моделей. Такое разделение необходимо из-за разных хостов, способов авторизации и ограничений Ozon.

Поддерживаемые эндпоинты:

• POST /api/client/token — client.performance.auth.get_token();
• подробный ответ со статусом и заголовками — get_token_detailed().
• GET /api/client/statistics/daily/json — client.performance.statistics.get_daily();
• подробный ответ со статусом и заголовками — get_daily_detailed().
• POST /api/client/statistics/json — client.performance.statistics.get();
• подробный ответ со статусом и заголовками — get_detailed().
• GET /api/client/statistics/{UUID} — client.performance.statistics.get_report_status();
• подробный ответ со статусом и заголовками — get_report_status_detailed().
• GET /api/client/statistics/report?UUID={uuid} — client.performance.statistics.download_report();
• подробный ответ со статусом и заголовками — download_report_detailed().
• POST /v3/product/list — client.seller.products.get_list();
• подробный ответ со статусом и заголовками — get_list_detailed();
• курсорный обход товаров — iter_list().
• POST /v3/product/info/list — client.seller.products.get_info_list();
• подробный ответ со статусом и заголовками — get_info_list_detailed().
• POST /v1/product/info/stocks-by-warehouse/fbo — client.seller.products.get_fbo_stocks_by_warehouse();
• подробный ответ со статусом и заголовками — get_fbo_stocks_by_warehouse_detailed();
• курсорный обход остатков — iter_fbo_stocks_by_warehouse().
• POST /v1/analytics/data — client.seller.analytics.get_data();
• подробный ответ со статусом и заголовками — get_data_detailed().
• POST /v1/finance/accrual/by-day — client.seller.finance.get_accruals_by_day();
• подробный ответ со статусом и заголовками — get_accruals_by_day_detailed();
• курсорный обход начислений — iter_accruals_by_day().
• POST /v1/report/placement/by-products/create — client.seller.reports.create_placement_by_products();
• подробный ответ со статусом и заголовками — create_placement_by_products_detailed().
• POST /v1/report/info — client.seller.reports.get_info();
• подробный ответ со статусом и заголовками — get_info_detailed().

import asyncio

from ozapi import AsyncOzonClient, SellerAPIKeyAuth


async def main() -> None:
    async with AsyncOzonClient(
        seller_auth=SellerAPIKeyAuth(
            client_id="your-client-id",
            api_key="your-api-key",
        )
    ) as client:
        response = await client.seller.products.get_list(
            {
                "filter": {"visibility": "ALL"},
                "last_id": "",
                "limit": 100,
            }
        )
        print(response["result"]["items"])

        async for product in client.seller.products.iter_list(
            {"filter": {"visibility": "VISIBLE"}, "limit": 1000},
            max_pages=10,
        ):
            print(product["product_id"])

        product_info = await client.seller.products.get_info_list(
            {"offer_id": ["offer-1", "offer-2"]}
        )
        print(product_info["items"])

        fbo_stocks = await client.seller.products.get_fbo_stocks_by_warehouse(
            {
                "limit": 1000,
                "offer_ids": ["offer-1", "offer-2"],
            }
        )
        print(fbo_stocks["products"])

        async for stock in client.seller.products.iter_fbo_stocks_by_warehouse(
            {"limit": 1000, "offer_ids": ["offer-1", "offer-2"]},
            max_pages=10,
        ):
            print(stock["sku"], stock["warehouse_id"], stock["present"])

        analytics = await client.seller.analytics.get_data(
            {
                "date_from": "2026-07-01",
                "date_to": "2026-07-20",
                "dimension": ["day"],
                "metrics": ["revenue", "ordered_units"],
                "filters": [],
                "sort": [{"key": "revenue", "order": "DESC"}],
                "limit": 1000,
                "offset": 0,
            }
        )
        print(analytics["result"]["data"])

        async for accrual in client.seller.finance.iter_accruals_by_day(
            {"date": "2026-07-20", "last_id": ""},
            max_pages=10,
        ):
            print(accrual["accrual_id"])

        report = await client.seller.reports.create_placement_by_products(
            {
                "date_from": "2026-06-01",
                "date_to": "2026-06-30",
            }
        )
        print(report["code"])

        report_info = await client.seller.reports.get_info({"code": report["code"]})
        print(report_info["result"]["status"])
        if report_info["result"]["status"] == "success":
            print(report_info["result"]["file"])


asyncio.run(main())

Performance API автоматически получает и кэширует bearer-токен:

import asyncio

from ozapi import AsyncOzonClient, PerformanceCredentials


async def main() -> None:
    async with AsyncOzonClient(
        performance_credentials=PerformanceCredentials(
            client_id="your-performance-client-id",
            client_secret="your-performance-client-secret",
        )
    ) as client:
        daily = await client.performance.statistics.get_daily(
            {
                "campaignIds": ["123456"],
                "dateFrom": "2026-07-01",
                "dateTo": "2026-07-20",
            }
        )
        print(daily)

        statistics = await client.performance.statistics.get(
            {
                "campaigns": ["123456"],
                "dateFrom": "2026-07-01",
                "dateTo": "2026-07-20",
                "groupBy": "DATE",
            }
        )
        print(statistics)

        report_uuid = "uuid-полученного-ранее-отчёта"
        report_status = await client.performance.statistics.get_report_status(
            report_uuid
        )
        if report_status["state"] == "OK":
            report_bytes = await client.performance.statistics.download_report(
                report_uuid
            )
            print(f"Получено байт: {len(report_bytes)}")


asyncio.run(main())

Для клиента нужно настроить хотя бы одно семейство API: передать seller_auth, performance_credentials или оба параметра. Учётные данные Performance API отправляются только в JSON-теле запроса токена; заголовки Seller API и Authorization в этот запрос не добавляются. Для методов статистики клиент сам получает bearer-токен, конкурентно-безопасно кэширует его до истечения срока и не более одного раза обновляет после ответа 401. Публичные get_token() и get_token_detailed() по-прежнему доступны для явного получения токена. Представление OzonResponse через repr намеренно не включает тело, чтобы случайно не показать токен или содержимое отчёта.

get_daily() сериализует каждый элемент campaignIds отдельным query-параметром. Если период не указан, Ozon возвращает последние семь дней. Для get() Ozon допускает не более десяти кампаний и период до 62 дней согласно сохранённой спецификации. Оба метода возвращают исходный JSON без runtime-преобразования и сохраняют неизвестные поля. Методы статуса работают с UUID отчёта, созданного асинхронным CSV-сценарием Performance API. download_report() возвращает исходные bytes: Content-Type из варианта *_detailed() позволяет отличить CSV от ZIP.

Для последовательного получения всех страниц используйте iter_list(), iter_fbo_stocks_by_warehouse() или iter_accruals_by_day(). Параметр max_pages позволяет явно ограничить число запросов. Метод get_info_list() принимает массивы offer_id, product_id и/или sku; суммарно в одном запросе можно передать не более 1000 товаров. Бета-метод get_fbo_stocks_by_warehouse() принимает offer_ids или skus, обязательный limit не больше 1000 и необязательный cursor. SKU в запросе передаются строками, как указано в сохранённой Seller Swagger-спецификации. Метод get_data() принимает период, список группировок dimension, до 14 метрик и limit от 1 до 1000. Поле offset позволяет вручную получать следующие страницы. Без Premium-подписки доступны только последние три месяца и ограниченный набор группировок и метрик; точный состав описан в сохранённой Seller Swagger-спецификации.

По умолчанию клиент автоматически соблюдает общий лимит Seller API в 50 запросов в секунду для одного Client-Id. Встроенный ограничитель координирует клиенты, потоки и асинхронные задачи только внутри одного процесса Python. Метод чтения безопасно повторяется не более двух раз после 429, временных 5xx, транспортных ошибок и тайм-аутов. Каждая попытка проходит через ограничитель. Для создания отчёта о стоимости размещения дополнительно действует лимит пять запросов в день. После 5xx, транспортной ошибки или тайм-аута создание отчёта автоматически не повторяется. Получение информации об отчёте использует общий лимит Seller API и безопасно повторяется при 429, временных 5xx, транспортных ошибках и тайм-аутах. Для данных аналитики действует отдельный лимит один запрос в минуту. Временные ошибки этого читающего endpoint повторяются не более двух раз, причём каждая попытка также проходит через минутный ограничитель. Для Performance API применяется документированный общий лимит 100 000 запросов в сутки на client_id, а для выгрузок статистики — более строгий лимит 2000 выгрузок за 24 часа с аккаунта. Одна кампания считается одной выгрузкой, поэтому клиент списывает отдельное разрешение для каждого явно переданного ID кампании. В документации Ozon также указано не более одной одновременной выгрузки с аккаунта и пяти с организации. Клиент сериализует выгрузки одного client_id внутри процесса. Встроенные ограничения координируют клиенты, потоки и асинхронные задачи только внутри одного процесса Python; лимит организации между разными client_id, процессами или машинами остаётся ответственностью внешнего общего backend.

Получение токена и читающие GET-методы безопасно повторяются не более двух раз после 429, временных 5xx, транспортных ошибок и тайм-аутов. POST /api/client/statistics/json после 5xx, транспортной ошибки или тайм-аута автоматически не повторяется, чтобы не создавать лишние выгрузки; ограниченные повторы после 429 сохраняются. Каждая попытка проходит через ограничитель.

Файлы в specs/upstream/ хранятся без ручных исправлений. При их обновлении дата загрузки и SHA-256 фиксируются в specs/README.md.