API основных модулей

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

Модуль Storage

Система хранения ключ-значение на базе SQLite, поддерживающая универсальные SQL-цепочечные запросы.

Основные операции

from ErisPulse import sdk

sdk.storage.set("key", "value")
value = sdk.storage.get("key", default_value)
keys = sdk.storage.keys()
sdk.storage.delete("key")

Массовые операции

sdk.storage.set_multi({"key1": "val1", "key2": "val2"})
values = sdk.storage.get_multi(["key1", "key2"])
sdk.storage.delete_multi(["key1", "key2"])

Транзакционные операции

with sdk.storage.transaction():
    sdk.storage.set("key1", "value1")
    sdk.storage.set("key2", "value2")

Доступ к свойствам

sdk.storage.my_key          # эквивалентно sdk.storage.get("my_key")
sdk.storage.my_key = "val"  # эквивалентно sdk.storage.set("my_key", "val")

SQL-цепочечные запросы

Модуль Storage предоставляет гибкий SQL-конструктор запросов в стиле цепочек вызовов, поддерживающий CRUD-операции для пользовательских таблиц.

sdk.storage.CreateTable("users", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "name": "TEXT NOT NULL",
})

sdk.storage.Table("users").Insert({"name": "Alice"}).Execute()
rows = sdk.storage.Table("users").Select("name").Where("id > ?", 0).Execute()

Полный API цепочечных запросов (Select/Insert/Update/Delete/Where/OrderBy/Limit, AlterTable, транзакции и др.) см. в разделе SQL-конструктор запросов.

Абстракция хранилища

StorageManager наследуется от абстрактного базового класса BaseStorage, что позволяет расширять поддержку других хранилищ (Redis, MySQL и др.).

from ErisPulse.Core.Bases.storage import BaseStorage, BaseQueryBuilder

Асинхронные интерфейсы

Модули Storage и Config предоставляют асинхронные методы (с префиксом a), которые можно безопасно вызывать в асинхронных обработчиках. Синхронные методы сохраняются без изменений, что не требует модификации существующего кода.

# Асинхронное хранение
value = await sdk.storage.aget("key")
await sdk.storage.aset("key", "value")
await sdk.storage.adelete("key")
keys = await sdk.storage.aget_all_keys()
await sdk.storage.aclear()

# Асинхронные массовые операции
values = await sdk.storage.aget_multi(["k1", "k2"])
await sdk.storage.aset_multi({"k1": "v1", "k2": "v2"})
await sdk.storage.adelete_multi(["k1", "k2"])

# Асинхронная конфигурация
value = await sdk.config.agetConfig("MyModule.key")
await sdk.config.asetConfig("MyModule.key", "value")
await sdk.config.aforce_save()
await sdk.config.areload()

Модуль Config

Управление конфигурационными файлами в формате TOML, поддержка ключевых путей, разделённых точками.

Обзор API

Примеры

config = sdk.config.getConfig("MyModule", {})
value = sdk.config.getConfig("MyModule.timeout", 30)

sdk.config.setConfig("MyModule", {"key": "value"})
sdk.config.setConfig("MyModule.timeout", 60, immediate=True)

setConfig по умолчанию использует отложенную запись (каждые 5 секунд происходит пакетное сохранение), установка immediate=True немедленно сохраняет изменения в конфигурационный файл. Изменения конфигурации запускают событие жизненного цикла config.set.

Модуль Logger

Модульная система логирования, основанная на Rich, поддерживает под-логгеры и контроль на уровне модуля.

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

sdk.logger.debug("Отладочная информация")
sdk.logger.info("Информация о работе")
sdk.logger.warning("Предупреждение")
sdk.logger.error("Ошибка")
sdk.logger.critical("Критическая ошибка")

Под-логгеры

child_logger = sdk.logger.get_child("MyModule")
child_logger.info("Лог подмодуля")

child_logger.get_child("utils")  # Поддержка вложенных логгеров

Контроль уровня логирования

sdk.logger.set_level("DEBUG")                          # Глобальный уровень
sdk.logger.set_module_level("MyModule", "DEBUG")       # Уровень на уровне модуля

# Поддерживаемые уровни (от низкого к высокому):
# TRACE, DEBUG, INFO, WARNING, ERROR, CRITICAL
# TRACE - самый низкий уровень, выводит подробную отладочную информацию (распространение событий, регистрация маршрутов и т.д.)
sdk.logger.set_level("TRACE")                          # Включить все логи

Подписка на логи (режим push)

Для модулей, таких как Dashboard, для получения структурированных логов в реальном времени, поддерживается фильтрация по уровню и отправка истории.

Явная подписка на логи низкого уровня: Уровень min_level подписчика может быть ниже глобального уровня логирования. В этом случае логи низкого уровня отправляются только соответствующему подписчику, не выводятся в консоль и не записываются в память, тем самым предотвращая загрязнение основного потока логов.

# Глобальный уровень INFO, но можно отдельно подписаться на DEBUG логи
@sdk.logger.handler("debug-tracer", min_level="DEBUG")
def on_debug(log_data: dict): ...

# Способ с декоратором
@sdk.logger.handler("my-handler", min_level="INFO")
def on_log(log_data: dict):
    # log_data = {
    #     "timestamp": "2026-06-29T22:00:00.123456",
    #     "level": "WARNING", "level_num": 30,
    #     "module": "ErisPulse.Core.adapter",
    #     "message": "Строгий режим:...",
    # }
    pass

# Прямой вызов
sdk.logger.handler("my-handler", min_level="INFO")(on_log)
sdk.logger.remove_handler("my-handler")

Контроль вывода

sdk.logger.set_output_file("app.log")
sdk.logger.save_logs("log.txt")
sdk.logger.get_logs("MyModule")
sdk.logger.set_memory_limit(1000)

Модуль адаптера

Менеджер адаптеров, отвечающий за регистрацию, запуск и остановку адаптеров для различных платформ.

Обзор API

События адаптера

@sdk.adapter.on("message")
async def handle_message(event):
    pass

@sdk.adapter.on("message", platform="yunhu")
async def handle_yunhu_message(event):
    pass

Запрос состояния бота

sdk.adapter.get_bot_info("telegram", "123456")
sdk.adapter.list_bots("telegram")
sdk.adapter.is_bot_online("telegram", "123456")
sdk.adapter.get_status_summary()

Полный список API управления адаптерами см. в API системы адаптеров.

Модуль

Менеджер модулей, отвечающий за регистрацию, загрузку и выгрузку плагинов.

Обзор API

Доступ к свойствам

module = sdk.module.get("ModuleName")
module = sdk.module.ModuleName
module = sdk.ModuleName  # Эквивалентный сокращённый способ

Вызовы между модулями (RPC)

# Протоколированный вызов: типизированные ошибки / автоматическое пробуждение ленивых модулей / привязка owner / семантика таймаута
result = await sdk.module.call("Chat", "get_history", session_id, n=20)

Различие между вызовом sdk.module.call() и прямым доступом к атрибутам sdk.module.Chat.get_history(...):

Служебный контракт (meta.services)

Поставщик службы объявляет белый список доступных методов в поле services метода get_meta() (аналогично commands), после чего доступ к вызовам сужается:

class ChatModule(BaseModule):
    @staticmethod
    def get_meta() -> ModuleMeta:
        return ModuleMeta(services=["get_history", "translate"])

    async def get_history(self, session_id, n=20): ...

• По умолчанию = разработчику не видно: если services не объявлен, доступны любые публичные методы (для обратной совместимости), методы с подчёркиванием всегда запрещены; основной контроль ограничений осуществляется на стороне пользователя через конфигурацию scope
• После объявления services: доступны только методы из белого списка, при попытке вызова метода за пределами списка выбрасывается ServiceNotProvidedError
• Ограничение для вызывающей стороны: scope.set_action("CallerModule", "call", deny="Chat.get_history")

Описание службы (description): services поддерживает словарную форму для описания каждой службы (поддерживает как строки, так и словари i18n), используется для каталога служб / описания точек вызова для ИИ:

return ModuleMeta(
    services=[
        "get_history",                              # Простая форма: описание автоматически берётся из первой строки docstring метода
        {"name": "translate", "description": "Перевод текста на указанный язык"},
        {"name": "summarize", "description": {"i18n": "Chat.meta.svc.summarize", "default": "Суммировать диалог"}},
    ],
)

Приоритет анализа описания: явное описание (i18n разрешается в текущем языке) > первая строка docstring метода > пустая строка.

Каталог служб (services)

sdk.module.services()
# {'Chat': [{'name': 'get_history', 'signature': '(session_id, n=20)',
#            'description': 'Получить историю сессии'}]}

sdk.module.services("Chat")  # Получить только информацию о конкретном модуле

В каталоге перечисляются только модули, явно объявившие meta.services; для каждой службы указываются строка сигнатуры метода и текст описания, что служит основой для MCP (экспозиции точек вызова для ИИ).

Направленная доставка событий относится к слою жизненного цикла: lifecycle.emit(event, data, to="ModuleName"), подробнее см. Межмодульная коммуникация.

Модуль Lifecycle

Система управления жизненным циклом на основе событий, обеспечивающая функции отправки и прослушивания событий.

Обзор API

Примеры

@sdk.lifecycle.on("module.init")
async def handle_module_init(event_data):
    print(f"Инициализация модуля: {event_data}")

@sdk.lifecycle.on("module")
async def handle_any_module_event(event_data):
    print(f"Событие модуля: {event_data}")

await sdk.lifecycle.emit("custom.event", {"key": "value"})

# Направленная доставка: событие будет доставлено только обработчикам, зарегистрированным в модуле Chat
await sdk.lifecycle.emit("message_received", {"text": "hi"}, to="Chat")

Полный список стандартных событий и подробное руководство см. в разделе Управление жизненным циклом.

Модуль Router

Менеджер маршрутизации HTTP/WebSocket, основанный на FastAPI + Uvicorn, поддерживает маршрутизацию с помощью декораторов, промежуточные слои, группировку, ограничение скорости, CORS.

Полная документация API маршрутизатора (декораторы маршрутов, WebSocket, промежуточные слои, ограничение скорости, CORS, заголовки безопасности и т.д.) доступна в разделе Маршрутизатор.

Краткий справочник

# HTTP маршрутизация
@sdk.router.get("MyModule", "/api")
async def handler(request: HttpRequest):
    return {"status": "ok"}

# WebSocket маршрутизация
@sdk.router.ws("MyModule", "/ws")
async def ws_handler(ws: WebSocketConnection):
    async for text in ws.iter_text():
        await ws.send_text(f"Echo: {text}")

# Группировка маршрутов
group = sdk.router.group("MyModule", prefix="/v1")
@group.get("/users")
async def list_users(request: HttpRequest):
    return {"users": []}

Модуль HTTP-клиента

Единый сетевой клиент, объединяющий HTTP-запросы, подключения WebSocket, управление пулом соединений, автоматическую повторную отправку, статистику запросов и интеграцию событий жизненного цикла.

Полная документация сетевого клиента (методы запросов, объекты ответов, клиент WebSocket, система исключений и т.д.) доступна в разделе Сетевой клиент.

Быстрая справка

from ErisPulse.Core import client

# HTTP-запрос
resp = await client.get("https://api.example.com/users")
data = await resp.json()

# WebSocket
ws = await client.ws_connect("wss://example.com/ws")
async for text in ws.iter_text():
    await ws.send_text(f"Echo: {text}")

SDK отладка

dump_state()

Экспорт текущего состояния работы фреймворка в виде снимка, используется для отладки и диагностики.

import json
state = sdk.dump_state()
print(json.dumps(state, indent=2, ensure_ascii=False, default=str))

Возвращаемая структура содержит состояние следующих подсистем:

Взаимодействие (Interaction)

Управление wait_reply (ожидание ответа) и сеансовыми взаимоисключающими арендами (sdk.interaction).

Часто используемые методы

# Напоминание сессии: если ответа нет в течение 5 минут, отправить напоминание, автоматически отменяется при ответе пользователя
reminder = event.remind(300, "Есть кто-нибудь?") 
reminder.cancel()  # ручная отмена

# Превышение времени: срабатывает по истечении срока (не отменяется при ответе)
event.escalate(1800, lambda e: notify_master("30 минут не обработано"))

# Ожидание нескольких путей: первый пришедший ответ имеет приоритет
which, reply = await event.select(
    event.expect(pattern="подтверждаю*", user="A"),
    event.expect(pattern="отказываюсь*", user="B"),
    timeout=60,
)

# Ожидание на уровне сессии: любой ответ в группе может быть принят
reply = await event.wait_reply(session=True, prompt="Кто-нибудь может помочь ответить?")

# Получение текущего владельца сессии (кто взаимодействует с пользователем)
owner = sdk.interaction.get_owner_of(event)

# Заявка на сеансовую взаимоисключающую аренду (возвращает None, если занят)
lease = sdk.interaction.acquire(event)
if lease:
    try:
        ...  # эксклюзивное взаимодействие
    finally:
        lease.release()

# Форма контекстного менеджера (выбрасывает SessionOccupiedError, если занят)
with sdk.interaction.hold(event) as lease:
    ...

# Статистика ожидания сессий
sdk.interaction.counts()  # {'waits': 2, 'leases': 1, 'timers': 3, 'owners': {'Chat': 3}}

При отключении модуля или адаптера все ожидания и таймеры автоматически отменяются (ожидающие немедленно получают None), а при совпадении ответа автоматически проверяется права доступа scope (если пользователь заблокирован или модуль отключен, ожидание завершается).

Система хранения переписки

Автоматическая запись и запрос потока последних сообщений для каждой сессии (sdk.transcript), служащий базой для модулей с памятью контекста, таких как AI-диалоги и предотвращение повторений.

Основные методы

# Удобный запрос (рекомендуется): последние 20 сообщений текущей сессии (включая пользователя и робота, по возрастанию времени)
messages = await event.history(20)
for m in messages:
    print(m["role"], ":", m["text"])

# API менеджера
sdk.transcript.append(event, "user", "текст")
sdk.transcript.get(event, n=20)
sdk.transcript.clear(event)

Конфигурация (ErisPulse.transcript): enabled (по умолчанию включено), max_per_session (максимум на сессию, по умолчанию 50), ttl_hours (время жизни по умолчанию, по умолчанию 168 часов). Данные хранятся в отдельной таблице SQLite, избыток или просроченные данные удаляются лениво.

Связанные документы

• API-системы событий - API-модуля Event
• API-системы адаптеров - API-управления адаптерами
• SQL-конструктор запросов - Полная документация SQL-цепного запроса
• Менеджер маршрутизации - Полная документация менеджера маршрутизации
• Сетевой клиент - Полная документация сетевого клиента
• Управление жизненным циклом - Полная документация жизненного цикла