基础示例

约 165 字小于 1 分钟

源码：examples/minimal_source_example.py

uv run examples/minimal_source_example.py

预期输出：

ready

"""不依赖外部服务的最小完整示例."""

import asyncio
from dataclasses import dataclass

from butterbot.app import BotApp, Event, RuntimeConfig
from butterbot.core.data import BaseDataMixin
from butterbot.core.source import BaseSource
from butterbot.core.types import BaseType


class TickType(BaseType):
    """示例事件类型."""

    ALL = "tick.all"
    READY = "tick.ready"


@dataclass
class TickData(BaseDataMixin):
    """示例事件数据."""

    message: str


class TickSource(BaseSource):
    """启动后发布一次事件，并负责回收自己的后台任务."""

    supported_types = TickType

    def __init__(self) -> None:
        super().__init__()
        self._task: asyncio.Task[None] | None = None

    async def on_start(self) -> None:
        self._task = asyncio.create_task(self._publish_once())

    async def _publish_once(self) -> None:
        event = Event(data=TickData(message="ready"), status=TickType.READY)
        await self.ctx.bus.publish(self.uuid, event)

    async def on_stop(self) -> None:
        task = self._task
        self._task = None
        if task is None:
            return
        if not task.done():
            task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass


async def main() -> None:
    """运行应用，收到事件后通过上下文管理器正常关闭."""
    app = BotApp(RuntimeConfig())
    source = app.add_source(TickSource)
    received = asyncio.Event()

    @app.subscribe(source.uuid, TickType.READY)
    async def handle_tick(event: Event[TickData]) -> None:
        print(event.data.message)
        received.set()

    async with app:
        await asyncio.wait_for(received.wait(), timeout=1.0)


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

验证范围

tests/test_examples.py::test_minimal_source_example_runs_and_exits_cleanly 使用当前 Python 解释器启动子进程，要求：

5 秒内退出；
退出码为 0；
标准输出为 ready。

该路径覆盖 asyncio.run()、BotApp 创建、Source 注册、Handler 类型关系、 Source task 所有权、async context manager 与正常关闭。

改造成真实 Source

替换 _publish_once() 为外部监听循环时：

保留 task 引用；
在 on_stop() 取消并等待；
每条原始输入先转换为 Data 和 BaseType；
不在 Source 中直接调用业务 Handler；
为取消、超时和连接异常增加测试。