Cluster · разработчикам

Объекты SDK

У каждого раздела SDK есть объекты для обычной работы: запись с вложенными полями, файл, загрузка, задача, сетевой запрос, ответ, соединение WSS, вызов модуля, договор обмена, входящий адрес и страница. Объект хранит своё состояние и параметры. Изменения применяются явным асинхронным методом либо при успешном выходе из async with там, где это предусмотрено.

Короткие методы из SDK.md сохранены. Объекты сообщений, текста, кнопок, клавиатур и альбомов описаны в BUILDERS.md.

Состав

Создание объекта само по себе не обращается к ядру. Сеть, файлы и другие службы работают через прежние проверки прав. Объект не получает доступ к путям сервера, сессии Telegram, токену бота или произвольному сокету.

Вложенные значения

Document работает в памяти, без базы и без прав. Строковый путь обозначает один ключ, кортеж или список задаёт вложенный путь. Точки в имени ключа не разбираются как разделители.

value = app.document({"profile": {"name": "User"}, "items": [3, 1, 2], "count": 0})
value.set(("profile", "name"), "New name")
value.rename(("profile", "name"), "title")
value.sort("items").append("items", 4)
value.move("items", 0, 2)
value.insert("items", 0, 0)
value.increment("count", 2)
name = value.get(("profile", "title"))

Методы изменения возвращают тот же объект. remove(path) удаляет ключ или элемент списка, set((), value) заменяет корень. get(path=(), default=...) возвращает копию значения. Без default отсутствующий путь вызывает обычную ошибку Python. append(path, *values) дополняет список, insert(path, index, value) вставляет элемент, move(path, index, target) переносит его. sort(path=(), key=None, reverse=False) сортирует список либо порядок ключей словаря.

values содержит локальное значение, dirty показывает отличие от последнего принятого состояния. reset() отменяет локальные изменения, accept() принимает их за исходное состояние. Для Record, Page и Settings состояние принимается автоматически после успешного сохранения. Вызывать у них accept() вручную для подтверждения записи не нужно.

Допустимы значения JSON: строки, числа, bool, None, списки и словари со строковыми ключами. Нечисловые NaN и бесконечность, произвольные Python-объекты и слишком глубокие структуры отклоняются.

Записи базы

record = await app.db.record("settings", {"count": 0, "items": []}).load()
record.increment("count").append("items", "example")
await record.save()

load() получает значение, наличие записи, срок и непрозрачную версию. record.exists отличает отсутствующую запись от сохранённого None. default используется только при отсутствии записи и не записывается автоматически при чтении.

save(ttl=...) сохраняет значение только при совпадении версии. Если другой обработчик, экземпляр или вызов уже изменил запись, ядро возвращает conflict. Локальные изменения при этом остаются в объекте. Загрузи актуальное состояние и реши, какие правки применить заново; автоматического объединения нет.

Без ttl сохраняется прежний срок записи. ttl=None явно снимает срок, целое число задаёт новый срок от текущего времени ядра. record.expires показывает абсолютное время окончания, None означает хранение без срока.

async with app.db.record("counter", {"count": 0}) as record:
    record.increment("count")

При успешном выходе изменённая запись сохраняется. Если внутри блока возникла ошибка, сохранения нет. Если сохранение получило конфликт, ошибка выходит из async with. Неизменённый объект не записывается. Для явного создания отсутствующей записи с default вызови save().

await record.delete() тоже сверяет версию и возвращает, существовала ли запись. Нужны db.read и db.write; загрузка требует только чтение. Версия относится к значению, наличию и сроку записи. Это проверка текущего состояния: история всех промежуточных изменений не хранится.

Разделы и пакет изменений

notes = app.db.scope("notes/")
await notes.set("first", {"text": "Example"})
record = await notes.record("first").load()

async with notes.batch() as batch:
    batch.set("second", {"text": "Another example"})
    batch.delete("first", revision=record.revision)

Префикс дописывается к каждому ключу. scope() можно вложить ещё раз; текущие module и slot сохраняются. Другой раздел того же юзербота требует db.full. Префикс является соглашением об именах, а не отдельной границей безопасности.

У Batch есть set(key, value, ttl=None, revision=None), delete(key, revision=None) и await commit(). Переданный revision добавляет проверку версии конкретного ключа. Без него значение заменяется без сравнения. Обычный batch.set без срока хранит запись бессрочно, как короткий app.db.set.

От 1 до 100 разных ключей применяются одной транзакцией. Повтор одного ключа в пакете отклоняется. Ошибка версии, типа, квоты или любого действия отменяет весь пакет. После успешного commit() повторно применить этот же объект нельзя. При ошибке внутри async with автоматического применения нет.

Весь запрос должен помещаться в предел сообщения SDK, исходно 1 MiB. Пакет не разбивается на несколько запросов: иначе он перестал бы быть атомарным. При проверке версий требуется также db.read; пакет без таких проверок может работать только с db.write.

from __future__ import annotations
from typing import TYPE_CHECKING

if TYPE_CHECKING: from router import Event

class Module:
    @loader.command("notes", access="owner")
    async def notes(self, ctx: Event) -> None:
        texts = []
        async for key, value in app.db.scope("notes/").items(limit=100):
            texts.append(f"{key}: {value}")
        await ctx.reply("\n".join(texts) or "Empty")

items(limit=100) проходит все порции. Ключи возвращаются без префикса раздела. Значение читается отдельно для каждого ключа, поэтому весь обход не является единым снимком базы. Записи могут измениться во время просмотра.

Файлы

file = await app.files.store("note.txt", b"Example", ttl=3600)
await file.rename("renamed.txt")
await file.retain(7200)
text = await file.text()
copy = await file.copy("copy.txt")
await copy.delete()

store загружает байты и возвращает File. Старый upload по-прежнему возвращает словарь. app.files(id) создаёт объект существующего файла, await file.load() получает метаданные. file.data содержит их, file.id идентификатор. В строковое представление объекта содержимое не включается.

rename(filename) меняет подпись файла. retain(ttl=None) меняет срок: число задаёт его заново от текущего времени, None снимает срок. Метаданные меняются с проверкой версии; устаревшая копия получает conflict. Уже истёкший файл таким способом не восстанавливается.

read(maximum=1048576) получает байты в память. Для больших файлов используй chunks(size=65536). Размер части можно уменьшить, например до 16384, если результат дальше передаётся функции другого модуля. read_at(offset=0, size=65536) читает одну часть с указанной позиции и возвращает байты. Размер части от 1 до 65536; позиция не может быть отрицательной. text(encoding="utf-8", maximum=1048576) декодирует содержимое, json(maximum=1048576) разбирает JSON. Ошибка декодирования или разбора возвращается вызывающему. copy(filename=None, ttl=None) копирует содержимое частями, проверяет сумму и создаёт новый ID. Копия занимает отдельное место и не получает чужие разрешения на чтение исходного файла.

Нужны files.read для чтения и files.write для создания, изменения и удаления. Чтение выбранного через конфигуратор чужого файла не даёт права его переименовывать. Доступ проверяется ядром при каждой операции.

file = await app.files.store("temporary.txt", b"Example", ttl=3600)
async with file:
    text = await file.text()

Такой блок удаляет файл при выходе. Если тело блока завершилось с ошибкой, неудача очистки не заменяет исходную ошибку. При потере связи или отзыве прав удалить файл может быть невозможно; для временных файлов задавай ttl.

app.files.iter(limit=100) является асинхронным обходом объектов File. Он продолжает чтение и после пустой порции с курсором. Неизвестный или повторяющийся курсор вызывает ошибку вместо бесконечного цикла.

Потоковая загрузка

async with app.files.writer("export.bin", size=expected_size, sha256=expected_sha256, ttl=3600) as writer:
    async for part in source:
        await writer.write(part)
file = writer.file

size задаётся до начала загрузки. sha256 необязателен, но при наличии проверяется ядром. write(data: bytes) принимает произвольную по размеру часть и сам делит её на блоки до 64 KiB. Позиция хранится в writer.offset; случайную позицию передавать не нужно.

Успешный выход вызывает finish() и сохраняет готовый объект в writer.file. Несовпадение размера или суммы, ошибка записи, исключение и отмена вызывают удаление незаконченной загрузки. open(), write(), finish() и abort() доступны и отдельно, если управление блоком неудобно. abort() не удаляет уже завершённый файл.

Операции одного загрузчика выполняются последовательно. Это не задаёт порядок двух частей, одновременно переданных через gather: передавай части в нужной последовательности.

Расписание

plan = app.tasks(self.save, values={"value": 5}).after(60).every(300).retry(2)
task = await plan.create()
await task.reschedule(new_timestamp)
await task.cancel()

handler принимает имя или связанный метод с @loader.task. values содержит поля обработчика. Для связанного метода можно передать args, как у кнопок: словарь либо позиционные значения в порядке объявления полей.

at(value) задаёт Unix-время или datetime с часовым поясом. after(seconds) задаёт задержку, которую ядро отсчитывает от своего текущего времени при создании. Значение 0 ставит задачу на ближайший проход расписания. Повтор создания с тем же ID не переносит срок заново. every(seconds) задаёт период целым числом секунд, retry(count) число повторов от 0 до 10. arguments(**values) дополняет параметры до создания.

Один объект плана имеет постоянный ID создания. Повторный create() возвращает ту же задачу, а не добавляет ещё одну. После создания меняй её через Task; правки плана больше не принимаются.

app.tasks.task(id) открывает существующую задачу. load() обновляет data, state и версию. edit(at=..., interval=..., retry=..., values=...) применяет изменения с проверкой версии, reschedule(at) меняет время, after(seconds) переносит запуск относительно текущего времени ядра. Завершённую или отменённую задачу можно поставить заново, если её обработчик и права остаются доступны. Выполняемую задачу сначала нужно отменить. Сохранённые полномочия создателя не заменяются полномочиями нового вызывающего.

cancel() отменяет задачу, delete() удаляет запись, wait(timeout=30, poll=0.1) ждёт одноразовый результат и поднимает ошибку при отмене или неудаче. Периодическая задача после успешного выполнения снова ожидает запуска; wait() не является способом дождаться одного её периода. Истечение ожидания не отменяет задачу.

app.tasks.iter(state=None) проходит задачи экземпляра с необязательным фильтром состояния. Все операции требуют tasks.create. Ограничения квоты, остановки среды при отмене выполняемой задачи и полномочий описаны в SDK.md.

HTTPS

request = app.web("https://example.org/items", method="POST")
request.query(page=1, tag=["one", "two"]).header("Accept", "application/json")
request.json({"name": "Example"})
response = await request.send()
data = response.raise_for_status().json()

query(**values) меняет параметры адреса с кодированием значений. Список создаёт несколько параметров с одним именем, None удаляет параметр. header(name, value) заменяет заголовок без учёта регистра имени. json(value), form(**values) и text(value, content_type=...) задают тело и его тип. clone() создаёт независимую копию параметров.

send() выполняет один запрос и возвращает Response. У ответа есть status, headers, ok, body, binary и content. json() разбирает JSON, text(encoding="utf-8") декодирует тело, header(name, default=None) ищет заголовок без учёта регистра. Для получения исходных байтов поставь request.binary=True перед отправкой. Без этого ядро возвращает текст с заменой повреждённых последовательностей.

Статус 4xx или 5xx сам по себе не является исключением. raise_for_status() проверяет его явно и возвращает тот же ответ. Короткие await app.web.get(url, **query) и await app.web.post(url, value) тоже возвращают Response. Старый request(...) сохраняет словарь результата.

session = app.web.session("https://example.org/api/", headers={"Accept": "application/json"})
response = await session.get("items", page=1)
await session.post("items", {"name": "Example"})

У Session общий начальный адрес и заголовки, у request(path="", method="GET") отдельные параметры запроса. session.header(name, value) меняет общие заголовки для следующих запросов. Переход на другую схему или имя сервера отклоняется, чтобы общие заголовки не ушли постороннему получателю. Автоматических перенаправлений и повторов нет.

Cookies и прокси остаются общими для экземпляра модуля, как в коротких методах. Создание Session не создаёт отдельный профиль cookies. Для изменения профиля есть app.web.profile(...) с прежними проверками.

await request.download(filename, maximum=10485760) получает файл в хранилище и возвращает File. Поддерживается простой GET без своих заголовков и тела. await request.upload(file) отправляет файл как тело, возвращает Response; метод должен быть POST, PUT или PATCH. Передаётся объект File либо его ID. Одновременно задать другое тело нельзя.

У запроса сохраняются ограничения HTTPS, разрешённых доменов, DNS, размера и заголовков из SDK.md. Строковое представление запроса не содержит URL, тела или заголовков, но это не делает безопасной явную запись этих значений в журнал.

WSS

async with app.web.socket("wss://example.org/events") as socket:
    await socket.send_json({"action": "subscribe"})
    value = await socket.receive_json()

open() создаёт соединение, close() закрывает. send(text) отправляет строку, send_json(value) сериализует JSON. receive(wait=10) возвращает строку, байты либо None после ожидания. receive_json разбирает полученное сообщение. Асинхронный обход async for value in socket пропускает пустые ожидания и завершается после локального close().

Успешный выход из блока закрывает соединение. При ошибке закрытие выполняется без замены исходной ошибки. Отзыв прав всё равно закрывает соединение на стороне ядра. Повторного открытия или автоматической отправки после обрыва нет. Ограничения сайтов, числа соединений, размера сообщений и доступных направлений передачи сохраняются.

Вызовы модулей

helper = app.modules("MODULE_ID", slot="main")
call = helper.function("calculate", value=5)
await call.start()
state = await call.poll()
result = await call.wait(timeout=30)

select(slot) возвращает другую ссылку на тот же модуль с выбранным экземпляром. function(name, **values) подготавливает вызов, arguments(**values) меняет его поля до отправки. start() ставит вызов в очередь один раз и сохраняет call.id. poll() возвращает состояние без ожидания. wait(timeout=30, poll=0.1) сам вызывает start(), если это ещё не сделано, затем ждёт результат.

После истечения ожидания тот же объект можно опросить или ждать дальше: повторная задача не создаётся. Если ответ на саму постановку потерян и ID неизвестен, объект не отправляет вызов повторно. Создание нового вызова в таком случае является отдельным решением автора.

Есть короткие await helper.call(function, values=None, timeout=30), await helper.emit(signal, values=None) и await app.modules.start(module, function, values=None, slot=None). Последний возвращает начатый объект Call.

Сохраняются modules.call, выбор зависимости, ограничения цепочки и текущие права всех её участников. Нет метода принудительной отмены выполняемой функции другого модуля. Истечение локального ожидания её не останавливает.

Обмен юзерботов

request = app.exchange(peer_id, module_id, "main")
request.use("event", "changes").direction("send").mode("permanent")
request.fields({"value": {"type": "integer", "required": True}})
agreement = await request.request()
await agreement.load()

use(kind, key) выбирает ресурс: data, text, file или event. Для файла можно передать объект File. direction принимает send или receive, mode значения permanent, once, group; для последнего можно задать group. fields принимает словарь либо объект Fields.

request() создаёт договор и возвращает Agreement. Повтор на том же объекте возвращает его же. После неизвестного результата отправки автоматического повторения нет. Получатель подтверждает договор обычным способом на сайте. SDK не одобряет запросы за него и не получает доступ к ресурсу до согласия.

У договора есть id, data, state, load(), transfer(value=None), revoke(), history(), events(after=0, limit=100) и clear(before). events проходит сохранённые порции с указанной позиции, это не бесконечная подписка. app.exchange.iter(limit=100) проходит договоры экземпляра. Для существующего ID есть app.exchange.agreement(id).

Все действия требуют exchange.request. Типы ресурсов, согласие, состояния, квоты и поведение при остановке сторон описаны в EXCHANGE.md.

Входящие запросы

endpoint = await app.requests(self.receive).create()
path = endpoint.path
headers = endpoint.headers
await endpoint.rotate()
await endpoint.revoke()

Обработчик должен быть объявлен через @loader.request. Можно передать его имя либо связанный метод. create() сохраняет адрес и ключ в объекте. Повтор на том же живом объекте не меняет ключ; rotate() делает это явно. Новый объект не умеет получить прежний ключ из ядра: у ядра хранится только хеш. Создание после перезапуска выдаст новый ключ.

headers возвращает Authorization: Bearer ..., path путь, token сам ключ. Строковое представление объекта ключ не содержит. Не сохраняй headers или token в общедоступной странице и не пиши их в журнал. При потере ответа на создание требуется явная смена ключа. Все вызовы требуют requests.manage.

Конфигурация и поля

fields = app.config.fields()
fields.add("enabled", app.config.field("boolean").default(True))
fields.add("count", app.config.field("integer").bounds(1, 100).default(5).when("enabled", True))
fields["count"].label({"ru": "Количество", "en": "Count"})

Field описывает одно поле с теми же типами, что словарь конфигурации. Методы label, default, bounds, choices, when, required меняют описание. У Fields есть add, доступ по ключу, remove, rename, move, sort и build. Переименование обновляет условия зависимых полей. Удалить поле, на которое ссылается условие, нельзя. Циклические условия отклоняются.

Готовый Fields можно присвоить Module.config, передать как fields декоратору или договору. SDK переводит его в словарь при регистрации. PageDefinition аналогично допустим в Module.pages. Описание в пакете должно совпадать с результатом build(). Конструктор не добавляет полей к уже установленной версии без обновления описания пакета.

view = await app.config.snapshot().load()
count = view.get("count", default=0)
changed = await view.changed()

У снимка есть values и revision. changed() проверяет версию в ядре, load() обновляет значения. Это локальный снимок конфигурации своего модуля; изменение словаря не меняет настройки ядра. Для чтения требуется config.read. Секретные значения своего модуля могут входить в снимок, поэтому не публикуй весь values.

Страницы

page = app.ui.define("main", {"ru": "Состояние", "en": "Status"})
page.add(app.ui.element("count", "metric", "Count"))
page.add(app.ui.element("items", "table", "Items", columns=["name", "value"]))
page.move("count", 0)

PageDefinition задаёт состав страницы в исходнике модуля. У него есть add, remove, rename, доступ по ID элемента, move, sort и build. Типы элементов те же, что в описании пакета: text, metric, table, button, form, tasks. Для кнопки и формы нужен объявленный action, для таблицы список columns. Element.label(...) меняет подпись перед регистрацией.

Значения опубликованной страницы меняются отдельно:

async with app.ui.page("main") as page:
    page.metric("count", 2)
    page.table("items").row(name="B", value=2).row(name="A", value=1).sort("name")

load() получает данные, версию и объявленные элементы. text(id, value) меняет текст, metric(id, value) показатель. table(id) возвращает объект таблицы с методами row(**values), cell(row, column, value), remove(row), move(row, target), sort(column=None, key=None, reverse=False) и clear(). rows() даёт локальный список строк. При сортировке по столбцу значения должны быть сравнимыми; для смешанных значений задай свой key.

save() сверяет версию и сохраняет все подготовленные данные. Параллельная правка через короткий app.ui.set тоже определяется как конфликт, если содержимое отличается. После ошибки локальные правки остаются. Успешный выход из async with сохраняет изменённую страницу, ошибка внутри блока оставляет ядро без изменений.

Состав и порядок элементов определяются установленным описанием. Через данные страницы нельзя добавить необъявленный элемент, сменить обработчик или расширить права. Для объекта страницы нужны interface.read и interface.write; старый set по-прежнему может использовать только запись.

Общие настройки и команды

async with app.core.settings_object() as settings:
    settings.set("prefix", ".")
    settings.set("notify_update", False)

commands = await app.core.commands_object("MODULE_ID/main").load()
commands.set("ping", "status")
await commands.save()

configuration_object(instance) работает с обычными полями конфигурации выбранного модуля. У всех трёх видов есть fields, values, revision, load, get, set, default(name), save и отмена локальных правок через reset(). Сохраняются только изменённые значения. Тип и границы проверяются до запроса, затем повторно ядром.

Права и происхождение не меняются: core.configure для общих настроек, modules.configure для конфигурации и команд. Требуется действующее прямое событие овнера в «Избранном» либо личном боте. Фоновая задача и чужой вызов не получают эту возможность. После завершения события объект нельзя использовать для обхода проверки. Конфликт версии не исправляется автоматической повторной записью.

app.core.iter_modules() проходит все доступные порции списка настраиваемых модулей в том же событии овнера.

Секретные поля и привязки файлов через эти объекты не меняются. В конфигурации других модулей значение секрета остаётся отметкой о наличии. Для изменения используй сайт.

Журнал

log = app.logs.bind(operation="export")
await log.info("Started", count=5)
async with log.timer("Export") as timer:
    await operation()

bind(**values) возвращает журнал с общими полями. Повторный bind создаёт дочерний объект, не меняя исходный. info, warn, error добавляют поля конкретной записи. Те же методы есть непосредственно у app.logs.

timer(message) измеряет время монотонными часами. При успехе пишет обычную запись с seconds, при ошибке запись с именем класса исключения. Текст исключения и traceback автоматически не включаются. Ошибка записи журнала при обработке исключения не заменяет исходное исключение. При обычном успешном выходе отказ журнала возвращается вызывающему.

Сохраняются logs.write, предел 4096 байт и ограничение частоты. Произвольные поля не являются разрешением писать пароли или токены. Переданные значения попадут в журнал; убирай секреты до вызова.

События

stream = app.stream("telegram").filter(lambda event: event.get("text", "").startswith("#"))
async for event in stream:
    await app.logs.info("Matching message", id=event["id"])

Первый next(wait=10) или шаг обхода начинает получение. События, пришедшие до этого, специально не накапливаются для нового подписчика. next возвращает одно подходящее событие либо None после ожидания. filter(predicate) задаёт синхронный фильтр, close() прекращает локальный обход. dropped содержит число событий, потерянных очередью ядра с момента получения этим объектом.

События одного экземпляра и источника получают из общей очереди. Два объекта не создают две независимые подписки: они будут забирать события друг у друга, как и при смешивании с app.events. Для независимых обработчиков используй декораторы роутера. Сохраняются права чтения, ограничения чатов и пользователей и проверка поколения подключения.

Ошибки и повторение

У настоящего app есть app.errors: SDKError, AccessDenied, Conflict, InvalidInput, InvalidState, QuotaExceeded, Unavailable, FloodWait. Все они наследуют RuntimeError; существующий except RuntimeError продолжает работать. code содержит код ядра или тип ошибки Telegram. У ограничения частоты доступно retry_after, когда сервис передал его.

class Module:
    async def save(self) -> bool:
        record = await app.db.record("counter", 0).load()
        record.increment(())
        try: await record.save()
        except app.errors.Conflict: return False
        return True

Ошибки локальной сборки и декодирования остаются обычными ValueError, TypeError, KeyError, IndexError, UnicodeError или ошибкой разбора JSON. Они не превращаются в сообщения об отказе ядра.

SDK не повторяет внешние действия автоматически. Потеря ответа может означать, что действие уже выполнено. Для задачи план сохраняет ID создания, для вызова модуля объект сохраняет известный ID результата. Для HTTPS и передачи данных повторение выбирает автор. Сохранение настроек и записей после конфликта требует нового чтения и осознанного применения правок.

Пример модуля

Исходник и описание пакета показывают совместную работу записи, полей конфигурации, страницы, файла, расписания и журнала. Зависимости из PyPI для объектов SDK не нужны. В STORE.md описана публикация пакета.

Составной пример SDK добавляет HTTPS, вызов зависимости, загрузку файла, кнопки, входящий запрос и расписание. Image Generator показывает передачу PNG между модулями частями без полного доступа к чужим файлам.