KVStore

KVStore — тонкая обёртка над KV-биндингом Cloudflare Workers. Автоматически сериализует/десериализует значения и обрабатывает ошибки.

from edgebot import KVStore

store = KVStore(env.MY_KV)

Конструктор

KVStore(kv: Any)

Параметр	Описание
kv	KV-неймспейс из env (например, env.USERS, env.CACHE)

Методы

async get(key)

Прочитать значение по ключу.

await store.get(key: str) -> Any

Возвращает:

• None — ключа нет
• dict / list — если значение является валидным JSON-объектом или массивом
• str — для остальных строковых значений

async get_with_metadata(key)

Прочитать значение вместе с метаданными.

await store.get_with_metadata(key: str) -> tuple[Any, dict | None]

Возвращает пару (value, metadata). Оба элемента могут быть None.

async exists(key)

Проверить наличие ключа.

await store.exists(key: str) -> bool

Заметка

Под капотом выполняет get(), т.е. читает и само значение.

async put(key, value, *, ttl, metadata)

Записать значение по ключу.

await store.put(
    key: str,
    value: Any,
    *,
    ttl: int | None = None,
    metadata: dict | None = None,
) -> None

Параметр	Описание
value	Автоматически сериализуется (см. правила ниже)
ttl	Время жизни в секундах; None — хранить вечно
metadata	Метаданные ключа, возвращаются в list() без чтения value

Правила сериализации value:

• bytes / bytearray / memoryview → бинарные данные
• str → как есть
• Всё остальное → json.dumps()

async delete(key)

Удалить ключ. Идемпотентно — не падает, если ключа не было.

await store.delete(key: str) -> None

async list(*, prefix, limit, cursor)

Перечислить одну страницу ключей (без чтения значений).

await store.list(
    *,
    prefix: str | None = None,
    limit: int = 1000,
    cursor: str | None = None,
) -> KVListResult

async iter_keys(*, prefix, cursor, batch)

Асинхронный итератор ключей с автоматической пагинацией.

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

Типы данных

KVKey

from edgebot import KVKey

KVKey(
    name: str,           # полное имя ключа
    expiration: int | None,  # unix-время истечения TTL
    metadata: dict | None,   # метаданные ключа
)

KVListResult

from edgebot import KVListResult

KVListResult(
    keys: list[KVKey],       # ключи текущей страницы
    cursor: str | None,      # курсор для следующей страницы
    list_complete: bool,     # True если больше страниц нет
)

Исключения

Все операции KV при ошибке бросают KVError:

from edgebot import KVError

try:
    value = await store.get("key")
except KVError as e:
    print(f"KV ошибка: {e}")

Пример

store = KVStore(env.SETTINGS)

# Сохранить JSON
await store.put("config:theme", {"dark": True, "lang": "ru"})

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

# С TTL (1 час)
await store.put("cache:result", data, ttl=3600)

# С метаданными
await store.put("user:123", user_data, metadata={"role": "admin"})

# Итерация по ключам
async for key in store.iter_keys(prefix="user:"):
    print(key.name)