Управление жизненным циклом
ErisPulse предоставляет единый механизм хуков/жизненного цикла, предназначенный для мониторинга состояния работы компонентов системы, а также реализации расширенных функций аудита, статистики и пользовательской логики.
Система поддерживает три способа триггеризации событий:
await lifecycle.emit("event", data)— упрощённая версия, передаёт любые данные (to="Owner"для направленной доставки)lifecycle.emit_sync("event", data)— синхронная версия (для не-асинхронных контекстов)await lifecycle.submit_event("event", ...)— совместимость со старыми версиями, автоматически формирует стандартный формат события
Механизм обработки событий
Регистрация обработчиков
from ErisPulse import sdk
# Регистрация с помощью декоратора
@sdk.lifecycle.on("module.load")
async def on_module_load(data):
print(f"Загрузка модуля: {data}")
# Программная регистрация
sdk.lifecycle.register("module.load", on_module_load, priority=10)
# Отмена регистрации
sdk.lifecycle.unregister("module.load", on_module_load)
#批量 отмена регистрации по владельцу (автоматически вызывается при выгрузке модуля/адаптера)
removed = sdk.lifecycle.unregister_by_owner("MyModule")
print(f"Удалено {removed} обработчиков жизненного цикла")
Приоритет
Обработчики поддерживают параметр priority, чем больше значение, тем раньше обработчик будет выполнен (аналогично загрузчику модулей):
@sdk.lifecycle.on("adapter.event.receive", priority=10) # Выполняется первым
async def first_handler(data):
pass
@sdk.lifecycle.on("adapter.event.receive", priority=0) # Выполняется позже
async def second_handler(data):
pass
События с точечной структурой
При срабатывании конкретного события также срабатывают и его родительские события:
- При срабатывании
module.loadтакже срабатываетmodule - При срабатывании
adapter.event.receiveтакже срабатываютadapter.eventиadapter
Шаблоны (wildcards)
Регистрация с помощью * позволяет поймать все события:
@sdk.lifecycle.on("*")
async def on_anything(data):
print(f"Получено событие: {data}")
Направленная передача событий (emit to=)
Note
Эта функция доступна начиная с ErisPulse 2.8.0+.
При указании параметра to в emit() событие будет направлено только обработчикам, зарегистрированным владельцем (владельцем по умолчанию считается модуль, который зарегистрировал обработчик в on_load), остальные модули и обработчики с шаблоном * не будут уведомлены.
# Отправка события: событие будет направлено только обработчикам, зарегистрированным в модуле Chat
await sdk.lifecycle.emit("message_received", {"text": "hi"}, to="Chat")
# Подписка на событие (внутри модуля Chat): обработчики, зарегистрированные в этом модуле, будут получать событие
@sdk.lifecycle.on("message_received")
async def on_message_received(data): ...
@sdk.lifecycle.on("message") # Родительские события также обрабатываются (фильтруются по владельцу)
async def on_any(data): ...
- Если у владельца нет зарегистрированных обработчиков → событие не будет обработано (можно заранее проверить с помощью
has_handlers()) - При
dataв виде словаря автоматически добавляется_trace_id(не перезаписывает уже существующее значение) emit_sync/submit_eventтакже поддерживают параметрto=- Трехуровневая модель взаимодействия между модулями (RPC / направленная / широковещательная) описана в Взаимодействие между модулями
Однократная регистрация (once)
Начиная с версии 2.7.0, обработчики, зарегистрированные через lifecycle.once(), автоматически отменяются после первого срабатывания, что подходит для одноразовых обработчиков, таких как "первичная готовность":
@sdk.lifecycle.once("core.init.complete")
async def on_first_ready(data):
print("Первичная готовность, дальнейшие срабатывания не будут происходить")
- Используется та же семантика приоритета, что и в
on()(чем больше значениеpriority, тем раньше обработчик будет выполнен) - Автоматическая отмена регистрации, без необходимости ручной отмены через
unregister - Поддержка как синхронных, так и асинхронных обработчиков
Проверка наличия слушателей (has_handlers)
Для ускорения выполнения в критических участках кода можно заранее проверить наличие слушателей с помощью has_handlers(), чтобы избежать ненужного перебора событий и задач:
if sdk.lifecycle.has_handlers("message.sending"):
await sdk.lifecycle.emit("message.sending", send_ctx)
- Проверка охватывает точное имя события, шаблон
*и родительские события - Возвращает
False, если слушателей нет, что позволяет безопасно пропуститьemit
Обзор точек остановки хука
Типичный цикл событий жизненного цикла сообщения от поступления из платформы в фреймворк до завершения обработки:
sequenceDiagram
participant P as Платформа
participant A as Адаптер
participant F as Ядро фреймворка
participant M as Обработчик модуля
P->>A: Пришло событие от платформы
A->>F: adapter.event.receive (самый ранний)
F->>F: event.pre_process (перед выполнением обработчика)
F->>M: Доставка в обработчик (команды/сообщения/уведомления и т.д.)
M->>M: command.matched / command.executed
M->>F: event.reply()
F->>F: message.sending (перед отправкой)
F->>A: DSL отправки
A->>P: Отправка на платформу
A->>F: message.sent (отправка завершена)
F->>F: adapter.event.dispatched (доставка завершена)
В фреймворке встроены следующие точки остановки хука, пользователь может слушать любую точку остановки для реализации пользовательской логики через @sdk.lifecycle.on().
Основная инициализация
| Имя хука | Точка остановки | Данные |
|---|---|---|
core.init.start |
Начало инициализации SDK | {} |
core.init.stage |
Начало этапа инициализации (выдается в фоне) | {"stage": str}; значения: discovery / adapter_register / adapter_start / module_register / module_init / adapter_start_deferred / router_start |
core.init.complete |
Завершение инициализации SDK | {"duration": float, "success": bool, "stages": {stage: float}, "adapters": {"enabled": [str], "disabled": [str]}, "modules": {"enabled": [str], "disabled": [str]}, "error": str(только при неудаче)} |
core.uninit.complete |
Завершение обратной инициализации SDK | {"duration": float, "success": bool, "adapters_closed": int, "modules_unloaded": int, "module_properties_cleared": int, "module_properties_to_clear": [str], "error": str(только при неудаче)} |
Пример: отображение прогресса запуска
@sdk.lifecycle.on("core.init.stage")
def show_stage(data):
print(f"[Запуск] Вход в этап: {data['stage']}")
Изменение конфигурации
| Имя хука | Точка остановки | Данные |
|---|---|---|
config.set |
Изменение конфигурационного параметра | {"key": str, "old_value": Any, "new_value": Any} |
config.updated |
Обнаружено изменение всей конфигурации после редактирования config.toml | {"old_config": dict, "new_config": dict, "config_file": str} |
Пример: аудит конфигурации
@sdk.lifecycle.on("config.set")
def audit_config(data):
print(f"[Аудит] {data['key']}: {data['old_value']} -> {data['new_value']}")
Жизненный цикл модуля
| Имя хука | Точка остановки | Данные |
|---|---|---|
module.register |
Регистрация класса модуля в менеджере | {"module_name": str, "success": bool} |
module.load |
Завершение загрузки модуля (успешное инстанцирование) | {"module_name": str, "success": bool} |
module.init |
Завершение инициализации модуля (включая ленивую загрузку) | {"module_name": str, "success": bool} |
module.unload |
Выгрузка модуля | {"module_name": str, "success": bool} |
module.reload |
Завершение горячей перезагрузки модуля (включая перезагрузку зависимостей) | {"module_name": str, "success": bool, "full": bool}; при полной перезагрузке (reload_all) module_name равно "All", в payload дополнительно присутствует "results": dict[str, bool] |
Жизненный цикл адаптера
| Имя хука | Точка остановки | Данные |
|---|---|---|
adapter.load |
Завершение регистрации адаптера | {"platform": str, "success": bool} |
adapter.start |
Запуск адаптера | {"platforms": [str]} |
adapter.status.change |
Изменение состояния адаптера | {"platform": str, "status": str, "retry_count": int, "error": str(только при неудаче)}; возможные значения status: starting / started / start_failed / stopping / stopped / stop_failed / skipped-dependency / disabled |
adapter.stop |
Остановка адаптера | {"platforms": [str]} |
adapter.stopped |
Завершение остановки адаптера | {"platforms": [str]} |
adapter.bot.online |
Онлайн бота | {"platform": str, "bot_id": str, "info": dict, "status": str} |
adapter.bot.offline |
Оффлайн бота | {"platform": str, "bot_id": str, "status": str} |
Прием и обработка событий
| Имя хука | Точка остановки | Данные |
|---|---|---|
adapter.event.receive |
Получено событие от внешней платформы (самый ранний) | {"platform": str, "event_type": str, "raw_event_type": str} |
adapter.event.blocked |
Промежуточное ПО отклонило событие (возвращает False, событие отбрасывается и не попадает в обработчики) |
{"middleware": str, "platform": str, "event_type": str, "detail_type": str, "event": dict, "_trace_id": str} |
adapter.event.dispatched |
Завершена доставка события | {"platform": str, "event_type": str, "raw_event_type": str, "onebot_handlers_count": int} |
event.pre_process |
Начало выполнения обработчика события | {"event_type": str, "platform": str, "detail_type": str} |
Пример: статистика событий
event_counter = {}
@sdk.lifecycle.on("adapter.event.receive")
def count_events(data):
platform = data["platform"]
event_counter[platform] = event_counter.get(platform, 0) + 1
@sdk.lifecycle.on("adapter.event.dispatched")
def log_unhandled(data):
if data["onebot_handlers_count"] == 0:
print(f"[Необработано] {data['platform']}/{data['event_type']}")
Отправка сообщения
| Имя хука | Точка остановки | Данные |
|---|---|---|
message.sending |
Сообщение готовится к отправке | {"platform": str, "method": str, "detail_type": str, "target_id": str, "bot_id": str} |
message.sent |
Сообщение успешно отправлено | {"platform": str, "method": str, "detail_type": str, "target_id": str, "bot_id": str} |
Пример: аудит отправки сообщений
@sdk.lifecycle.on("message.sending")
def log_sending(data):
print(f"[Отправка] -> {data['platform']}/{data['detail_type']}/{data['target_id']} через {data['method']}")
Командная система
| Имя хука | Точка остановки | Данные |
|---|---|---|
command.matched |
Команда соответствует шаблону и готова к выполнению | {"command": str, "args": list[str], "platform": str, "user_id": str} |
command.executed |
Команда успешно выполнена | {"command": str, "args": list[str], "platform": str, "user_id": str, "success": bool, "error": str(только при неудаче)} |
Пример: статистика команд
@sdk.lifecycle.on("command.matched")
def count_commands(data):
print(f"[Команда] /{data['command']} от {data['user_id']}@{data['platform']}")
HTTP-маршрутизация
| Имя хука | Точка остановки | Данные |
|---|---|---|
server.request |
Получен HTTP-запрос | {"method": str, "path": str, "client_ip": str} |
server.response |
Отправлен HTTP-ответ | {"method": str, "path": str, "status_code": int, "client_ip": str} |
Пример: логирование HTTP-запросов
@sdk.lifecycle.on("server.response")
def log_http(data):
print(f"[HTTP] {data['method']} {data['path']} -> {data['status_code']}")
WebSocket
| Имя хука | Точка остановки | Данные |
|---|---|---|
server.start |
Запуск сервера маршрутизации | {"base_url": str, "host": str, "port": int, "success": bool, "error": str(только при неудаче)} |
server.stop |
Остановка сервера маршрутизации | {} |
server.websocket.connect |
Установлено WebSocket-соединение | {"path": str, "module_name": str, "client_ip": str} |
server.websocket.disconnect |
WebSocket-соединение разорвано | {"path": str, "module_name": str, "reason": str, "error": str(только при аномалии)} |
Пример: мониторинг WebSocket-соединений
@sdk.lifecycle.on("server.websocket.connect")
def on_ws_connect(data):
print(f"[WS] Подключение: {data['path']} от {data['client_ip']}")
@sdk.lifecycle.on("server.websocket.disconnect")
def on_ws_disconnect(data):
print(f"[WS] Отключение: {data['path']} ({data['reason']})")
Состояние подключения к хранилищу
Создание, сбой и восстановление пула подключений к хранилищу (все события отправляются в фоне, не блокируют операции с хранилищем):
| Имя хука | Точка остановки | Данные |
|---|---|---|
storage.ready |
Пул подключений к хранилищу готов (первое успешное создание пула в каждом цикле событий) | {"backend": str} |
storage.unreachable |
Повторные попытки подключения исчерпаны, вступает период охлаждения (в течение которого операции быстро завершаются с ошибкой) | {"backend": str, "error": str, "cooldown": float} |
storage.recovered |
Охлаждение закончено, повторное подключение успешно, хранилище снова доступно | {"backend": str} |
Пример: предупреждение о сбое хранилища
@sdk.lifecycle.on("storage.unreachable")
def alert_storage_down(data):
print(f"[Предупреждение] Хранилище {data['backend']} недоступно: {data['error']}, автоматическое повторное подключение через {data['cooldown']} секунд")
@sdk.lifecycle.on("storage.recovered")
def notify_storage_back(data):
print(f"[Восстановление] Хранилище {data['backend']} снова доступно")
HTTP-клиент
События запросов и подключений клиента sdk.client (все события отправляются в фоне):
| Имя хука | Точка остановки | Данные |
|---|---|---|
client.request.success |
HTTP-запрос успешно выполнен | {"method": str, "url": str, "status": int, "elapsed": float} |
client.request.failed |
HTTP-запрос не выполнен после исчерпания попыток повтора | {"method": str, "url": str, "error": str, "attempts": int, "elapsed": float} |
client.ws.connect |
Установлено WebSocket-соединение | {"url": str} |
Международная локализация
| Имя хука | Точка остановки | Данные |
|---|---|---|
i18n.language.changed |
Смена языка фреймворка (i18n.set_language) |
{"language": str, "previous": str} |
Стандартное определение событий
STANDARD_EVENTS = {
"core": ["init.start", "init.stage", "init.complete", "uninit.complete"],
"module": ["load", "init", "unload", "register", "reload"],
"adapter": [
"load", "start", "status.change", "stop", "stopped",
"event.receive", "event.dispatched",
"bot.online", "bot.offline",
],
"server": [
"start", "stop",
"request", "response",
"websocket.connect", "websocket.disconnect",
],
"event": ["pre_process"],
"message": ["sending", "sent"],
"command": ["matched", "executed"],
"config": ["set", "updated"],
"storage": ["ready", "unreachable", "recovered"],
"client": ["request.success", "request.failed", "ws.connect"],
"i18n": ["language.changed"],
}
Полная справочная документация API
Регистрация и отмена
| Метод | Описание |
|---|---|
@lifecycle.on(event, *, priority=0) |
Регистрация обработчика декоратором |
lifecycle.register(event, handler, *, priority=0) |
Программная регистрация |
lifecycle.unregister(event, handler=None) |
Отмена регистрации (если handler=None, отменяются все обработчики события) |
Триггеризация
| Метод | Описание |
|---|---|
await lifecycle.emit(event, data=None, *, to=None) |
Асинхронный триггер, обработчики выполняются параллельно (не блокируют друг друга, возвращается, когда все завершены), возвращаемые значения по приоритету перезаписывают data; при to триггер направлен на владельца |
lifecycle.fire(event, data=None, *, to=None) |
Фоновый триггер (бросить в корзину): обработчики выполняются в фоновых задачах, не ждут, без возвращаемого значения; при отсутствии слушателей — нулевые затраты. Подходит для частых горячих путей и чисто наблюдательных событий; для последовательных триггеров (например, config.set) используйте emit |
lifecycle.emit_sync(event, data=None, *, to=None) |
Синхронный триггер, асинхронные обработчики запускаются через create_task |
await lifecycle.submit_event(event_type, *, source, msg, data, to=None, background=False) |
Совместимость со старыми версиями, автоматически формирует стандартный формат события; при background=True используется фоновый триггер fire |
Утилиты
| Метод | Описание |
|---|---|
lifecycle.start_timer(timer_id) |
Начать отсчёт времени |
lifecycle.get_duration(timer_id) |
Получить прошедшее время (в секундах) |
lifecycle.stop_timer(timer_id) |
Остановить отсчёт и вернуть прошедшее время |
lifecycle.list_hooks() |
Вывести все зарегистрированные хуки и количество обработчиков |
lifecycle.clear() |
Очистить все обработчики и таймеры |
Пример использования в модуле
from ErisPulse.Core.Bases import BaseModule
from ErisPulse import sdk
class Main(BaseModule):
async def on_load(self, event):
# Простая статистика сообщений
self.msg_count = 0
@sdk.lifecycle.on("adapter.event.receive")
async def count(data):
if data["event_type"] == "message":
self.msg_count += 1
# Мониторинг всех команд
@sdk.lifecycle.on("command.matched")
async def log_cmd(data):
sdk.logger.info(f"Команда выполнена: /{data['command']} от {data['user_id']}")
# Аудит изменений конфигурации
@sdk.lifecycle.on("config.set")
def audit(data):
sdk.logger.info(f"Изменение конфигурации: {data['key']} = {data['new_value']}")
Принадлежность фоновых задач и автоматическая отмена
Note
Эта функция доступна начиная с ErisPulse 2.8.0+.
Фоновые задачи, созданные модулем, если не отменены в on_unload, будут удерживать ссылку на self, что приведёт к невозможности сборки мусора модуля (остатки после горячей перезагрузки). Фреймворк предоставляет следующие механизмы:
self.spawn(coro)(рекомендуется в модуле): задача автоматически привязывается к имени модуля, при выгрузке модуля фреймворк автоматически отменяет незавершённые задачи и записывает предупреждение послеon_unloadspawn_background(coro)(вErisPulse.runtime): автоматически захватывает контекстowner_scope;cancel_owner_tasks(owner)отменяет задачи по принадлежности,cancel_all_background_tasks()используется для отмены всех фоновых задач вsdk.uninit()- Адаптеры: при остановке отменяются фоновые задачи по платформе
async def on_load(self, event):
# Рекомендуется: фоновые задачи использовать self.spawn(), фреймворк автоматически отменяет их при выгрузке
self.spawn(self._poll())
async def on_unload(self, event):
# В сценариях с точным контролем всё ещё рекомендуется отменять и ждать завершения
if self._poll_task:
self._poll_task.cancel()
await asyncio.gather(self._poll_task, return_exceptions=True)
async def _poll(self):
while True:
await asyncio.sleep(60)
...
Important
Фреймворк принудительно отменяет задачи (cancel_owner_tasks), это происходит после возврата из on_unload. Поэтому задачи, требующие корректного завершения (flush буферов, сохранение состояния, закрытие соединений), обязательно должны быть отменены и завершены в on_unload — не полагайтесь на отмену фреймворком. Фреймворк гарантирует только отсутствие задач, удерживающих self, но не гарантирует корректное завершение. Задачи, требующие ожидания результата, следует вызывать напрямую, а не передавать в фоновую задачу.
Примечания
- Обработчики могут быть синхронными или асинхронными: система автоматически определяет и правильно вызывает
- Передача данных: в режиме
emit()возвращаемое обработчиком значение отличное отNoneизменяет данные, передаваемые последующим обработчикам - Назначение имен событий: рекомендуется использовать точечную структуру для удобства использования родительских слушателей
- Изоляция ошибок: ошибка в одном обработчике не влияет на выполнение других обработчиков
- Ограничения синхронного триггера: в
emit_sync()асинхронные обработчики запускаются в фоне, возвращаемые значения не передаются - Очистка жизненного цикла: при вызове
sdk.uninit()все зарегистрированные обработчики и таймеры будут очищены - Приоритет загрузки: если необходимо прослушивать события на раннем этапе инициализации, рекомендуется установить высокий приоритет и отключить ленивую загрузку
Связанная документация
- Руководство по разработке модулей — подробности о методах жизненного цикла модуля
- Лучшие практики — рекомендации по использованию событий жизненного цикла