honker 深度解析：SQLite 上的 PostgreSQL NOTIFY/LISTEN 语义 — 事务型外盒模式实战与 AI Agent 消息队列架构

import honker

# 打开（或创建）数据库
db = honker.open("app.db")

# 创建名为 "emails" 的队列
emails = db.queue("emails")

# === 生产者：入队任务 ===
emails.enqueue({
    "to": "alice@example.com",
    "subject": "Welcome!",
    "body": "Hello Alice"
})
emails.enqueue({
    "to": "bob@example.com",
    "subject": "Order Confirmed",
    "body": "Your order #42 is confirmed"
})

# === 消费者：消费任务 ===
async for job in emails.claim("worker-1"):
    print(f"Processing job: {job.payload}")
    # 处理邮件发送...
    job.ack()  # 确认完成
    # 如果 worker 崩溃未 ack，任务会重新入队

import honker

db = honker.open("app.db")

# 创建持久化事件流
stream = db.stream("user-events")

# === 发布事件（在事务中）===
with db.transaction() as tx:
    tx.execute(
        "UPDATE users SET name=? WHERE id=?",
        ["New Name", 42]
    )
    # 只有事务提交成功，事件才会发布
    stream.publish(
        {"user_id": 42, "change": "name", "new_value": "New Name"},
        tx=tx
    )

# === 消费者：订阅事件流 ===
async for event in stream.subscribe(consumer="dashboard"):
    # 实时推送到浏览器
    await push_to_browser(event)
    # 事件持久化，重启后不会丢失

# ✅ 原子操作：数据写入和消息入队在同一个事务中
with db.transaction() as tx:
    db.execute("INSERT INTO users ...", tx=tx)
    db.queue("emails").enqueue({"to": user.email}, tx=tx)
# 事务要么全部成功，要么全部回滚

import honker
import json

db = honker.open("shop.db")

# 1. 创建表
db.execute("""
    CREATE TABLE IF NOT EXISTS orders (
        id INTEGER PRIMARY KEY,
        product TEXT,
        status TEXT DEFAULT 'pending',
        created_at DATETIME DEFAULT CURRENT_TIMESTAMP
    )
""")

# 2. 创建触发器（在 Python 中注册 honker 函数）
# 注：实际触发器需在 honker 扩展加载后创建

# 3. 创建订单流
order_stream = db.stream("order-events")

# 4. 创建订单（会自动触发流发布）
with db.transaction() as tx:
    tx.execute(
        "INSERT INTO orders (product, status) VALUES (?, ?)",
        ["MacBook Pro", "pending"]
    )
    # 触发器会自动发布事件到流
    # 如果没有触发器，手动发布：
    # order_stream.publish(
    #     {"action": "order_created", "product": "MacBook Pro"},
    #     tx=tx
    # )

# 5. 消费订单事件
async for event in order_stream.subscribe(consumer="order-processor"):
    data = json.loads(event)
    print(f"处理订单事件: {data}")
    # 可以触发 AI Agent 处理：
    # - 发送确认邮件
    # - 检查库存
    # - 更新推荐模型

import honker
import asyncio

db = honker.open("agent.db")
agent_queue = db.queue("agent-tasks")

# Agent 主循环：派发任务
async def dispatch_agent_task(task_type: str, payload: dict):
    agent_queue.enqueue({
        "type": task_type,
        "payload": payload,
        "priority": payload.get("priority", 5),
        "created_at": time.time()
    })

# Worker：处理任务
async def process_agent_tasks():
    async for job in agent_queue.claim("agent-worker-1"):
        task = job.payload
        if task["type"] == "generate_report":
            report = await ai_generate_report(task["payload"])
            # 将结果写回数据库
            job.ack()

# 记录 Agent 的每个决策
decision_stream = db.stream("agent-decisions")

with db.transaction() as tx:
    # 保存决策到数据库
    tx.execute(
        "INSERT INTO decisions (agent_id, action, context) VALUES (?, ?, ?)",
        [agent_id, action, json.dumps(context)]
    )
    # 同时发布到事件流（用于实时监控/调试）
    decision_stream.publish({
        "agent_id": agent_id,
        "action": action,
        "timestamp": time.time(),
        "context": context
    }, tx=tx)