FFI и утилиты

Модуль edgebot.ffi

Мост между JavaScript-рантаймом Cloudflare Workers (Pyodide) и Python.

to_js_object(value)

Конвертирует Python dict/list в нативный JS Object/Array (рекурсивно).

from edgebot import to_js_object

options = to_js_object({
    "method": "POST",
    "headers": {"Content-Type": "application/json"},
    "body": '{"key": "value"}',
})

Заметка

Нужно там, где Cloudflare API требует именно Object, а не Map (KV options, D1, R2 и т.д.).

to_py(value)

Конвертирует JsProxy обратно в Python с нормализацией nullish-значений.

from edgebot import to_py

result = to_py(js_value)

Правила:

• None, JS null, JS undefined → Python None
• JsProxy с .to_py() → результат конвертации
• Python-значения (str, int, dict) → возвращаются как есть

---

Модуль edgebot.utils.http

sleep(ms)

Асинхронная задержка через JS setTimeout (для Pyodide).

from edgebot import sleep

await sleep(1000)  # пауза 1 секунда

fetch(url, options, …)

Обёртка над js.fetch с автоматическими ретраями при временных ошибках.

from edgebot.utils.http import fetch

resp = await fetch(url, options, max_retries=3)

Параметр	По умолчанию	Описание
url	—	URL запроса
options	—	Опции fetch (через to_js_object)
max_retries	3	Максимум попыток
retry_on_status	{408, 429, 500, 502, 503, 504}	HTTP-статусы для retry

Поведение:

• Успешный ответ (resp.ok) возвращается сразу
• Статус не в retry_on_status — возвращается без retry
• 429 (Rate Limit) — ожидание по заголовку Retry-After
• Сетевые ошибки — retry с логированием

---

Модуль edgebot.utils.log

Структурированное логирование для Cloudflare Workers. Сообщения формируются как JSON:

{"msg": "...", "src": "edgebot.xxx", "key": "value"}

Уровни

from edgebot.utils import log

log.debug("Диагностика", source="edgebot.bot", user_id=123)
log.info("Операция завершена", source="edgebot.kv")
log.warn("Сообщение не изменилось", source="edgebot.ui")
log.error("KV недоступен", source="edgebot.storage", key="user:123")

Функция	Уровень в Workers Logs	Когда использовать
log.debug()	debug	Диагностика (скрыто в production)
log.info()	info	Нормальный ход работы
log.warn()	warn	Штатно, но требует внимания
log.error()	error	Операция не выполнилась

Все функции принимают:

• message: str — основное сообщение
• source: str — источник (по умолчанию "edgebot")
• **fields — произвольные поля для структурированного лога

---

Функция now()

Текущее UTC-время в ISO-8601 с суффиксом Z.

from edgebot.utils import now

timestamp = now()  # "2026-04-20T12:34:56Z"

Используется UserRegistry для created_at / updated_at.

---

Исключения

Полная иерархия исключений SDK:

EdgeBotError                  # Корень иерархии
├── UIError                   # Ошибки UI-движка
│   ├── KeyboardNotFound      # Ключ не найден в KeyboardRegistry
│   ├── PromptNotFound        # Ключ не найден в PromptRegistry
│   └── DuplicatePrefixError  # Два компонента с одним __prefix__
└── StorageError              # Ошибки хранилищ
    ├── KVError               # Ошибки KV-операций
    └── UsersError            # Ошибки реестра пользователей
        └── UserAlreadyExists # create() на существующий user_id

from edgebot import EdgeBotError

try:
    await bot.process_update(update)
except EdgeBotError as e:
    # Ловит любую ошибку SDK
    print(f"EdgeBot error: {e}")