aedb

aedb — Python SDK для демонов платформы. Публичный интерфейс позволяет подключить демон к NATS, принимать события и задачи, отвечать на RPC и обрабатывать длительные задачи JetStream.

Требуется Python 3.14.

python -m pip install aedb

Настройка демона

export DAEMON_NATS_HOST='nats://localhost:4222'
export DAEMON_NATS_SEED='SU...'

DAEMON_NATS_HOST — адрес NATS; DAEMON_NATS_SEED — секретный NKey seed демона. Оба значения обязательны. Не записывайте seed в исходный код. Таймауты подключения и переподключения можно задать через DAEMON_NATS_TIMEOUT и DAEMON_NATS_RECONNECT_TIMEOUT (по умолчанию 3 секунды). DaemonSettings() читает эти переменные при создании объекта. При необходимости можно передать NATSSettings(host=..., seed=...) явно.

Для первой активации демона укажите также DAEMON_JOIN_HOST — базовый URL Join-сервиса — и вызовите await daemon.join(token) с одноразовым токеном до запуска соединения. Join не вызывается автоматически; при следующих запусках уже активированного демона токен не нужен.

Приём и отправка событий

import asyncio

from aedb.contracts import Event
from aedb.engine.daemons import Daemon, DaemonSettings


daemon = Daemon(DaemonSettings())


@daemon.listen.events("demo.events")
async def on_event(event: Event) -> None:
    print(event.payload)


@daemon.listen.rpc("demo.echo")
async def echo(event: Event) -> Event:
    return Event(task_id=event.task_id, payload=event.payload)


async def main() -> None:
    async with daemon:
        await daemon.action.event("demo.events", {"text": "hello"})
        answer = await daemon.action.rpc("demo.echo", {"text": "ping"})
        print(answer.payload)


asyncio.run(main())

async with daemon: открывает NATS-соединение и подписки, а при выходе закрывает их. Если управление жизненным циклом нужно вручную, используйте await daemon.start() и await daemon.close(). Обработчики можно зарегистрировать до подключения. Для обмена сообщениями нужны права NATS на соответствующие subjects.

Event — JSON-конверт с id, необязательным task_id, meta, timestamp и словарём payload. Методы daemon.action.event, task и rpc принимают как Event, так и словарь payload. rpc возвращает Event.

Обработчик получает Event и может быть синхронной или асинхронной функцией. Для task и rpc результат отправляется только при наличии reply-subject и ненулевого результата. Для action.task ответ следует принимать отдельным обработчиком listen.events(reply).

Длительные задачи JetStream

@daemon.listen.long_task(...) принимает сообщения из заранее созданного durable consumer с explicit ack и фильтром по subject. Асинхронный обработчик возвращает Event с тем же task_id, что у запроса, либо словарь payload. SDK публикует ответ в reply_to и подтверждает входящее сообщение после успешной публикации. При ошибке запрашивается повторная доставка.

from aedb.contracts import Event


@daemon.listen.long_task(
    "demo.jobs.request",
    stream="DEMO_JOBS",
    consumer="worker",
    reply_to="demo.jobs.answer",
    progress_every=20.0,
    retry_delay=5.0,
)
async def process(event: Event) -> Event:
    return Event(task_id=event.task_id, payload={"result": event.payload})


# Выполнять после запуска демона:
# ack = await daemon.action.long_task(
#     "demo.jobs.request", Event(task_id="job-1", payload={"value": 42})
# )

progress_every должен быть меньше AckWait consumer. action.long_task возвращает подтверждение публикации JetStream, но не ждёт результата задачи. Результат поступает отдельно на reply_to.

Команда aedb engine key

Команда создаёт NKey-пару или выводит публичный ключ из существующего seed:

aedb engine key
aedb engine key --type account
aedb engine key --save ./worker
aedb engine key --seed "$(cat ./worker.seed)"
aedb engine key --seed "$(cat ./worker.seed)" --save ./worker-public

По умолчанию создаётся пара типа user. --type принимает user, account, server, operator или cluster. --save BASE сохраняет seed и публичный ключ в BASE.seed и BASE.pub с правами 0600. При --seed SEED новая пара не создаётся: выводится только публичный ключ, а с --save сохраняется только файл BASE.pub.