async-uds-api 0.1.7

Неофициальный асинхронный Python-клиент для UDS Partner API

async-uds-api

Асинхронный Python-клиент для публичного UDS Partner API v2.

Установка

pip install async-uds-api

Быстрый старт

import asyncio
from async_uds_api import UDSClient


async def main() -> None:
    client = UDSClient(
        company_id="123456",
        api_key="your-api-key",
    )

    async with client:
        # Получение настроек компании
        settings = await client.settings.get()
        print(settings.name, settings.currency)

        # Список клиентов
        customers = await client.customers.list(max=10, offset=0)
        for customer in customers.rows:
            print(customer.display_name, customer.phone)


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

Возможности

• Асинхронный HTTP-клиент на базе httpx.AsyncClient
• Авторизация через Basic Auth (companyId:apiKey)
• Автоматические заголовки X-Origin-Request-Id и X-Timestamp
• Pydantic-модели для валидации данных
• Типизация — полная поддержка статических анализаторов
• Обработка ошибок — иерархия исключений с детальной информацией

API

Settings

settings = await client.settings.get()

Customers

# Список клиентов
customers = await client.customers.list(max=100, offset=0)

# Поиск клиента по коду/телефону/UID
result = await client.customers.find(code="ABC123", total=1000.0)

# Получение клиента по ID
customer = await client.customers.get(customer_id=12345)

# Теги клиента
tags = await client.customers.get_tags(customer_id=12345)
await client.customers.set_tags(customer_id=12345, tag_ids=[1, 2, 3])

Operations

from async_uds_api.models import CreateOperation

# Список операций
operations = await client.operations.list(max=100)

# Создание операции (покупка/начисление)
operation = await client.operations.create(CreateOperation(...))

# Получение операции по ID
operation = await client.operations.get(operation_id=12345)

# Возврат операции
refunded = await client.operations.refund(operation_id=12345)

# Расчёт покупки
calc_result = await client.operations.calc(calc_request)

# Начисление бонусов
await client.operations.reward(reward_request)

# Создание ваучера
voucher = await client.operations.create_voucher(voucher)

Tags

# Список тегов компании
tags = await client.tags.list()

Goods

from async_uds_api.models import GoodsDetailed

# Список товаров
goods = await client.goods.list(max=100)

# Создание товара
new_goods = await client.goods.create(GoodsDetailed(name="Товар", ...))

# Получение по ID
item = await client.goods.get(goods_id=123)

# Обновление
updated = await client.goods.update(goods_id=123, goods=GoodsDetailed(...))

# Удаление
await client.goods.delete(goods_id=123)

# Работа с externalId
item = await client.goods.external.get(external_id="ext-123")
updated = await client.goods.external.update(external_id="ext-123", goods=...)
await client.goods.external.delete(external_id="ext-123")

Images

# Загрузка изображения из файла
image_id = await client.images.upload("/path/to/image.jpg")

# Загрузка из URL
image_id = await client.images.upload("https://example.com/image.png")

# Загрузка из байтов
image_id = await client.images.upload(image_bytes, content_type="image/png")

# Получение URL для загрузки
upload_url = await client.images.get_upload_url("image/jpeg")

Источником может быть путь в файловой системе, http(s)-URL или байты. Строка со схемой http или https скачивается по сети, любая другая строка интерпретируется как путь в файловой системе.

Goods Orders

from async_uds_api import GoodsOrderUpdate, GoodsOrderUpdateStatus

# Получение заказа
order = await client.goods_orders.get(order_id=123)

# Обновление заказа (позиции, доставка)
await client.goods_orders.update(order_id=123, body=GoodsOrderUpdate(...))

# Смена статуса заказа
await client.goods_orders.change_status(
    order_id=123, status=GoodsOrderUpdateStatus.READY
)

# Отмена заказа
await client.goods_orders.cancel(order_id=123)

# Завершение заказа (создаёт транзакцию)
result = await client.goods_orders.complete(order_id=123)
result.transaction.id  # ID созданной транзакции
result.order  # GoodsOrderDetailed

# Генерация кода оплаты
code_info = await client.goods_orders.generate_code(order_id=123)

Логирование

Библиотека пишет в логгер async_uds_api и по умолчанию ничего не выводит (NullHandler). Чтобы увидеть сообщения, настройте стандартный logging:

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(message)s",
)
logging.getLogger("async_uds_api").setLevel(logging.INFO)

Вывод:

GET /customers [max=50 cursor=abc] [X-Origin-Request-Id=6e1c89d3-...] [X-Timestamp=2026-07-18T19:59:54.981185+00:00]
GET /customers/find [phone=***4567] [X-Origin-Request-Id=6e1c89d3-...] [X-Timestamp=2026-07-18T19:59:54.981185+00:00]
GET /customers/find -> 200 OK in 0.312s

Маскирование персональных данных

В лог попадают все query-параметры запроса. Маскируются только phone, uid и code — они сокращаются до последних четырёх символов и никогда не попадают в лог целиком. Остальные параметры (например, max, cursor, offset) выводятся в логе без изменений — это помогает при разборе инцидентов.

Текст сообщения об ошибке, который UDS возвращает в ответе, может содержать эхо phone/uid/code — иногда в нормализованном виде, отличном от того, что было передано в запросе. Библиотека не пытается вычищать чужой текст регулярными выражениями: она просто никогда его не логирует. Событие uds.error не содержит поля message, а str(exc) у UDSAPIError — это безопасная сводка вида 400 for GET /customers/find [errorCode=badRequest].

В атрибут exc.message попадает только поле message из корректно разобранного JSON-объекта ответа — это документированное поле UDS API, его можно осознанно прочитать и, при необходимости, залогировать самому. Любое другое тело ответа (plain text или HTML-страница от прокси, WAF или CDN — такие страницы часто печатают запрошенный URI вместе с незамаскированным phone) в exc.message не попадает: вместо него используется та же безопасная сводка, что и при пустом теле (500 for GET /customers/find):

except UDSAPIError as e:
    print(e)          # 400 for GET /customers/find [errorCode=badRequest]
    print(e.message)  # поле message из JSON UDS, может содержать ПДн

То же относится и к httpx.HTTPStatusError, который остаётся в __cause__ у ошибок API-запросов: его собственный текст содержит полный URL запроса вместе с query-строкой, поэтому он переписывается на сводку вида 400 for GET /customers/find — без query-строки, а значит без phone/uid/code. Тип исключения и его .response не меняются. Благодаря этому телефон не появляется и в полном traceback, который печатает logging.exception.

Это правило касается только запросов к API UDS. URL в путях загрузки изображений (presigned-ссылки и адрес источника) не маскируются: они короткоживущие, а без полного URL непонятно, какой именно объект не загрузился. Сообщения и traceback этих путей содержат исходный URL целиком.

mask_value и mask_params экспортируются из async_uds_api — ими можно пользоваться и вне библиотеки, например в собственных логах.

По умолчанию UDSClient выставляет логгеру httpx уровень WARNING: на уровне INFO httpx печатает полный URL запроса вместе с query-строкой, то есть незамаскированный телефон. Отключить это поведение можно так:

client = UDSClient(company_id="...", api_key="...", silence_httpx_log=False)

Учтите, что при silence_httpx_log=False номера телефонов, uid и коды клиентов будут утекать в лог через URL запросов httpx.

Свой логгер

UDSClient принимает любой объект с методами debug/info/warning/error, которые получают имя события и поля через **kwargs. structlog и loguru подходят без обёртки:

import structlog

client = UDSClient(
    company_id="...",
    api_key="...",
    logger=structlog.get_logger(),
)

Также можно передать обычный logging.Logger или logging.LoggerAdapter — оба автоматически оборачиваются в StdlibLoggerAdapter, так что классический формат сообщений (см. пример вывода выше) сохраняется и для LoggerAdapter:

import logging

client = UDSClient(
    company_id="...",
    api_key="...",
    logger=logging.LoggerAdapter(logging.getLogger("myapp"), {}),
)

Если переданный объект не является ни Logger/LoggerAdapter, ни объектом с методами debug/info/warning/error, UDSClient бросает TypeError уже в конструкторе — до того, как логирование сломает первый же запрос.

События и их поля:

Событие	Уровень	Поля
uds.request	INFO	method, path, params, request_id, timestamp
uds.response	INFO	method, path, status, elapsed
uds.error	ERROR	method, path, status, elapsed, error_code
uds.retry	WARNING	method, path, attempt
uds.image.*	DEBUG/INFO/ERROR	зависит от события

События uds.image.* пишут URL как есть: поле url у uds.image.download_start, uds.image.download_done и uds.image.download_failed, поле source у uds.image.upload_start_source, а также тексты UDSImageDownloadError и UDSImageUploadError содержат полный URL. Это осознанный выбор: presigned-ссылки короткоживущие, а https://cdn.example.com/*** не говорит, какой объект не загрузился. Если такие URL не должны попадать в ваш лог, отфильтруйте эти события на стороне хендлера.

При стандартном логгере эти поля доступны хендлерам через record.uds — словарь с исходными значениями, удобный для JSON-форматтеров. Атрибут uds присутствует на всех записях, которые библиотека пишет через StdlibLoggerAdapter, включая события async_uds_api.api.images. Читать его всё равно стоит защищённо: getattr(record, "uds", None).

Диагностика логирования

Логирование никогда не должно ронять запрос: если пользовательский обработчик логов бросает исключение, StdlibLoggerAdapter по умолчанию молча его глотает. Чтобы увидеть это исключение при отладке, выставьте переменную окружения ASYNC_UDS_API_DEBUG_LOGGING в любое непустое значение — тогда исключение из обработчика будет пробрасываться наружу:

export ASYNC_UDS_API_DEBUG_LOGGING=1

Webhooks

is_valid = client.verify_webhook_signature(
    request_id=request.headers.get("X-RequestId"),
    timestamp=request.headers.get("X-Timestamp"),
    signature=request.headers.get("X-Signature"),
)

Обработка ошибок

from async_uds_api import (
    UDSClientError,
    UDSAPIError,
    UDSBadRequestError,
    UDSUnauthorizedError,
    UDSForbiddenError,
    UDSNotFoundError,
    UDSUnexpectedError,
    UDSImageError,
)

try:
    customer = await client.customers.get(999999)
except UDSNotFoundError as e:
    print(f"Не найдено: {e.message}")
except UDSAPIError as e:
    print(f"API ошибка: {e.status_code}, {e.error_code}")
except UDSClientError as e:
    print(f"Ошибка клиента: {e}")

str(e) у UDSAPIError — это безопасная сводка (404 for GET /customers/999999), пригодная для логирования. Текст, который вернул сервер, лежит в e.message и может содержать персональные данные: читайте его осознанно и не пишите в лог не подумав. Подробнее — в разделе Маскирование персональных данных.

Требования

• Python >= 3.10
• httpx >= 0.27.0
• pydantic >= 2.0.0
• aiofiles >= 23.0.0

Разработка

# Установка зависимостей для разработки
uv sync --dev

# Запуск тестов
uv run pytest tests/

# Линтинг
uv run ruff check async_uds_api/ tests/

# Форматирование
uv run ruff format async_uds_api/ tests/

# Проверка типов
uv run mypy

Лицензия

MIT