UserRegistry

UserRegistry — высокоуровневая обёртка для хранения данных пользователей в KV. Автоматически проставляет created_at / updated_at.

from edgebot import UserRegistry

users = UserRegistry(env.USERS)

Конструктор

UserRegistry(kv: Any, *, prefix: str = "user:")

Параметр	Описание
kv	KV-биндинг из env или готовый KVStore
prefix	Префикс ключей в неймспейсе (по умолчанию "user:")

Ключ в KV: {prefix}{user_id}, например "user:123456".

Методы

async get(user_id)

Прочитать запись пользователя.

await users.get(user_id: int) -> dict | None

async exists(user_id)

Проверить наличие записи.

await users.exists(user_id: int) -> bool

async create(user_id, data, *, ttl, metadata)

Создать новую запись. Автоматически добавляет created_at и updated_at.

await users.create(
    user_id: int,
    data: dict,
    *,
    ttl: int | None = None,
    metadata: dict | None = None,
) -> dict

Raises	Условие
UserAlreadyExists	Запись с таким user_id уже есть

async upsert(user_id, data, *, ttl, metadata)

Создать или полностью заменить запись. Сохраняет прежний created_at, обновляет updated_at.

await users.upsert(
    user_id: int,
    data: dict,
    *,
    ttl: int | None = None,
    metadata: dict | None = None,
) -> dict

async update(user_id, data, *, ttl, metadata)

Частично обновить поля (merge верхнего уровня). Обновляет updated_at.

await users.update(
    user_id: int,
    data: dict,
    *,
    ttl: int | None = None,
    metadata: dict | None = None,
) -> dict | None

Возвращает None, если записи нет.

async delete(user_id)

Удалить запись. Идемпотентно.

await users.delete(user_id: int) -> None

async iter_ids(*, batch)

Асинхронно итерировать user_id всех записей без чтения значений.

async for uid in users.iter_ids():
    print(uid)

async items(*, batch)

Итерировать пары (user_id, data) с чтением значения каждого ключа.

async for uid, data in users.items():
    print(uid, data["username"])

Заметка

items() делает отдельный get() для каждого ключа — дороже, чем iter_ids().

Служебные поля

UserRegistry автоматически добавляет к записям:

Поле	Формат	Описание
created_at	ISO-8601 UTC ("2026-04-20T12:34:56Z")	Время создания
updated_at	ISO-8601 UTC	Время последнего обновления

Исключения

from edgebot import UserAlreadyExists, UsersError

try:
    await users.create(123, {"name": "John"})
except UserAlreadyExists:
    print("Пользователь уже существует")

Пример

from edgebot import UserRegistry, Context

users = UserRegistry(env.USERS)

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

    if await users.exists(user_id):
        user = await users.get(user_id)
        await ctx.send(f"С возвращением, {user['name']}!")
    else:
        user = await users.create(user_id, {
            "name": ctx.from_user.get("first_name", "—"),
            "username": ctx.from_user.get("username"),
            "is_active": True,
        })
        await ctx.send(f"Добро пожаловать, {user['name']}!")