Хранение данных в KV

Cloudflare KV — key-value хранилище, встроенное в Workers. EdgeBot предоставляет две обёртки: низкоуровневый KVStore и высокоуровневый UserRegistry.

Подготовка

1. Создайте KV-неймспейс

npx wrangler kv namespace create USERS

2. Добавьте биндинг в wrangler.toml

[[kv_namespaces]]
binding = "USERS"
id = "YOUR_NAMESPACE_ID"

3. Подключите к боту

from edgebot import Bot, UserRegistry

def build_bot(env) -> Bot:
    users = UserRegistry(env.USERS)
    return Bot(env.BOT_TOKEN).with_users(users)

Регистрация пользователя

@bot.on_command("/start")
async def start(ctx: Context):
    user_id = ctx.from_user["id"]

    if await ctx.users.exists(user_id):
        await ctx.send("С возвращением!")
        return

    await ctx.users.create(user_id, {
        "username": ctx.from_user.get("username"),
        "first_name": ctx.from_user.get("first_name"),
        "settings": {"notifications": True},
    })
    await ctx.send("Добро пожаловать!")

Обновление данных

Частичное обновление (merge)

await ctx.users.update(user_id, {
    "settings": {"notifications": False},
})
# Остальные поля (username, first_name) не затрагиваются

Полная замена (upsert)

await ctx.users.upsert(user_id, {
    "username": "new_username",
    "first_name": "Новое имя",
    "settings": {"notifications": True},
})
# Сохраняет прежний created_at, обновляет updated_at

Произвольные KV-данные

Для данных, не привязанных к пользователям, используйте KVStore напрямую:

from edgebot import KVStore

store = KVStore(env.SETTINGS)

# Сохранить
await store.put("config:lang", "ru")
await store.put("config:limits", {"max_users": 100, "max_requests": 50})

# Прочитать
lang = await store.get("config:lang")       # "ru"
limits = await store.get("config:limits")   # {"max_users": 100, ...}

# С TTL (кэш на 1 час)
await store.put("cache:api_result", data, ttl=3600)

# Удалить
await store.delete("cache:api_result")

Итерация по записям

Только ID (быстро)

async for user_id in ctx.users.iter_ids():
    print(user_id)

С данными (медленнее)

async for user_id, data in ctx.users.items():
    if data.get("is_active"):
        print(f"{user_id}: {data['username']}")

Низкоуровневая итерация KV

store = KVStore(env.MY_KV)

async for key in store.iter_keys(prefix="cache:"):
    print(key.name, key.metadata)