API

API

API

API(
    token,
    token_owner=UNKNOWN,
    version=__vk_api_version__,
    requests_url="https://api.vk.ru/method/",
    requests_session=None,
    json_parser=None,
    cache_table=None,
    proxies=None,
    max_retries=5,
    retry_initial_delay=1.0,
    retry_max_delay=30.0,
)

Bases: SessionContainerMixin

Source code in vkflow\api.py

def __init__(
    self,
    token: str,
    token_owner: TokenOwner = TokenOwner.UNKNOWN,
    version: str = __vk_api_version__,
    requests_url: str = "https://api.vk.ru/method/",
    requests_session: aiohttp.ClientSession | None = None,
    json_parser: BaseJSONParser | None = None,
    cache_table: cachetools.Cache | None = None,
    proxies: list[str] | None = None,
    max_retries: int = 5,
    retry_initial_delay: float = 1.0,
    retry_max_delay: float = 30.0,
):
    SessionContainerMixin.__init__(self, requests_session=requests_session, json_parser=json_parser)

    if token.startswith("$"):
        self._token = os.environ[token[1:]]
    else:
        self._token = token

    self._version = version
    self._token_owner = token_owner
    self._owner_schema = None
    self._requests_url = requests_url
    self._proxies = proxies
    self._cache_table = cache_table or cachetools.TTLCache(ttl=7200, maxsize=2**12)

    self._method_name = ""
    self._last_request_timestamp = 0.0
    self._use_cache = False

    self._stable_request_params = {
        "access_token": self._token,
        "v": self._version,
    }

    self._max_retries = max_retries
    self._retry_initial_delay = retry_initial_delay
    self._retry_max_delay = retry_max_delay
    self._rate_limit_lock = asyncio.Lock()

    self._update_requests_delay()

method async

method(method_name, **request_params)

Выполняет необходимый API запрос с нужным методом и параметрами. Вызов метода поддерживает конвертацию из snake_case в camelCase.

Перед вызовом этого метода может быть вызван .use_cache() для включения возможности кэш-логики запроса

Каждый передаваемый параметр проходит специальный этап конвертации перед передачей в запрос по следующему принципу:

• Все элементы списков, кортежей и множеств проходят конвертацию рекурсивно и объединяются в строку через ,
• Все словари автоматически дампятся в JSON-строку установленным JSON-парсером
• Все True/False значения становятся 1 и 0 соответственно (требуется для aiohttp)
• Если переданный объект имплементирует класс APISerializableMixin, вызывается соответствующий метод класса для конвертации в желаемое значение

К параметрам автоматически добавляются access_token (ключ доступа) и v (версия API), переданные при инициализации, но каждый из этих полей может быть задан вручную для конкретного запроса. Например, необходимо вызвать метод с другой версией API или передать другой токен.

Parameters:

Name	Type	Description	Default
method_name	str	Имя вызываемого метода API	required
request_params	Any	Параметры, принимаемые методом, которые описаны в документации API.	{}

Returns:

Type	Description
Any	Пришедший от API ответ.

Raises:

Type	Description
VKAPIError	В случае ошибки, пришедшей от некорректного вызова запроса.

Source code in vkflow\api.py

async def method(self, method_name: str, **request_params: typing.Any) -> typing.Any:
    """
    Выполняет необходимый API запрос с нужным методом и параметрами.
    Вызов метода поддерживает конвертацию из snake_case в camelCase.

    Перед вызовом этого метода может быть вызван `.use_cache()` для
    включения возможности кэш-логики запроса

    Каждый передаваемый параметр проходит специальный этап конвертации перед
    передачей в запрос по следующему принципу:

    * Все элементы списков, кортежей и множеств проходят конвертацию рекурсивно и
        объединяются в строку через `,`
    * Все словари автоматически дампятся в JSON-строку установленным JSON-парсером
    * Все True/False значения становятся 1 и 0 соответственно (требуется для aiohttp)
    * Если переданный объект имплементирует класс `APISerializableMixin`,
        вызывается соответствующий метод класса для конвертации в желаемое
        значение

    К параметрам автоматически добавляются `access_token` (ключ доступа) и `v` (версия API),
    переданные при инициализации, но каждый из этих полей может быть задан вручную для
    конкретного запроса. Например, необходимо вызвать метод с другой версией API
    или передать другой токен.


    Arguments:
        method_name: Имя вызываемого метода API
        request_params: Параметры, принимаемые методом, которые описаны в документации API.

    Returns:
        Пришедший от API ответ.

    Raises:
        VKAPIError: В случае ошибки, пришедшей от некорректного вызова запроса.
    """
    use_cache = self._use_cache
    self._use_cache = False

    return await self._make_api_request(
        method_name=method_name,
        request_params=request_params,
        use_cache=use_cache,
    )

execute async

execute(*code)

Исполняет API метод execute с переданным VKScript-кодом.

Parameters:

Name	Type	Description	Default
code	str | CallMethod	VKScript код	()

Returns:

Type	Description
Any	Пришедший ответ от API

Raises:

Type	Description
VKAPIError	В случае ошибки, пришедшей от некорректного вызова запроса.

Source code in vkflow\api.py

async def execute(self, *code: str | CallMethod) -> typing.Any:
    """
    Исполняет API метод `execute` с переданным VKScript-кодом.

    Arguments:
        code: VKScript код

    Returns:
        Пришедший ответ от API

    Raises:
        VKAPIError: В случае ошибки, пришедшей от некорректного вызова запроса.
    """
    if not isinstance(code[0], str):
        code = "return [{}];".format(", ".join(call.to_execute() for call in code))

    return await self.method("execute", code=code)

use_cache

use_cache()

Включает кэширование для следующего запроса. Кэширование выключается автоматически, т.е. кэширование будет использовано только для первого следующего выполняемого API запроса.

Включение кэширования подразумевает, что следующий запрос будет занесен в специальную кэш-таблицу. Ключ кэша привязывается к имени вызываемого метода и переданным параметрам, а значение -- к ответу API. Если запрос с таким именем метода и такими параметрами уже был выполнен когда-то, то вместо отправки запроса будет возвращено значение из кэш-таблицы

Если необходимо передать свою собственную имплементацию кэш-таблицы, укажите соответствующий инстанс при инициализации объекта в поле cache_table. По умолчанию используется TTL-алгоритм.

Returns:

Type	Description
API	Тот же самый инстанс API, готовый к кэшированному запросу

Source code in vkflow\api.py

def use_cache(self) -> API:
    """
    Включает кэширование для следующего запроса.
    Кэширование выключается автоматически, т.е.
    кэширование будет использовано только для первого следующего
    выполняемого API запроса.

    Включение кэширования подразумевает, что следующий запрос
    будет занесен в специальную кэш-таблицу. Ключ кэша
    привязывается к имени вызываемого метода и переданным параметрам,
    а значение -- к ответу API. Если запрос с таким именем метода
    и такими параметрами уже был выполнен когда-то, то вместо
    отправки запроса будет возвращено значение из кэш-таблицы

    Если необходимо передать свою собственную имплементацию
    кэш-таблицы, укажите соответствующий инстанс при инициализации объекта
    в поле `cache_table`. По умолчанию используется TTL-алгоритм.

    Returns:
        Тот же самый инстанс API, готовый к кэшированному запросу
    """
    self._use_cache = True
    return self

define_token_owner async

define_token_owner()

Позволяет определить владельца токена: группа или пользователь.

Метод использует кэширование, поэтому в своем коде можно смело каждый раз вызывать этот метод, не боясь лишних исполняемых запросов

Владелец токена будет определен автоматически после первого выполненного запроса для определения задержки, если token_owner поле не было установленно вручную при инициализации объекта

Returns:

Type	Description
TokenOwner	Возвращает словарь, первый элемент которого TokenOwner значение,
Page	указывающее, группа это или пользователь, а в второй -- сама схема объекта
tuple[TokenOwner, Page]	сущности пользователя/группы, обернутая соответствующим враппером

:rtype:

Source code in vkflow\api.py

async def define_token_owner(self) -> tuple[TokenOwner, Page]:
    """
    Позволяет определить владельца токена: группа или пользователь.

    Метод использует кэширование, поэтому в своем коде
    можно смело каждый раз вызывать этот метод, не боясь лишних
    исполняемых запросов

    Владелец токена будет определен автоматически после первого выполненного
    запроса для определения задержки, если `token_owner` поле
    не было установленно вручную при инициализации объекта

    Returns:
        Возвращает словарь, первый элемент которого TokenOwner значение,
        указывающее, группа это или пользователь, а в второй -- сама схема объекта
        сущности пользователя/группы, обернутая соответствующим враппером
    :rtype:
    """
    if self._token_owner != TokenOwner.UNKNOWN and self._owner_schema is not None:
        return self._token_owner, self._owner_schema

    owner_schema = await self.use_cache().method("users.get")

    if owner_schema and isinstance(owner_schema, list) and len(owner_schema) > 0:
        self._owner_schema = User(owner_schema[0])
        self._token_owner = TokenOwner.USER

    else:
        owner_schema = await self.use_cache().method("groups.get_by_id")

        if owner_schema and isinstance(owner_schema, (list, dict)):
            group_data = None

            if isinstance(owner_schema, list) and len(owner_schema) > 0:
                group_data = owner_schema[0]

            elif isinstance(owner_schema, dict):
                if (
                    "groups" in owner_schema
                    and isinstance(owner_schema["groups"], list)
                    and len(owner_schema["groups"]) > 0
                ):
                    group_data = owner_schema["groups"][0]
                elif "id" in owner_schema:
                    group_data = owner_schema
                else:
                    raise ValueError(
                        f"groups.get_by_id returned invalid response: {owner_schema}. "
                        "Expected group object with 'id' field or dict with 'groups' key. "
                        "Please check your access token."
                    )

            if group_data is None or not isinstance(group_data, dict) or "id" not in group_data:
                raise ValueError(
                    f"groups.get_by_id returned invalid group data: {group_data}. "
                    "Expected dict with 'id' field."
                )

            self._owner_schema = Group(group_data)
            self._token_owner = TokenOwner.GROUP

        else:
            raise ValueError(
                "Unable to determine token owner. "
                "Both users.get and groups.get_by_id failed. "
                "Please check your access token."
            )

    self._update_requests_delay()
    return self._token_owner, self._owner_schema

upload_photos_to_message async

upload_photos_to_message(*photos, peer_id=0)

Загружает фотографию в сообщения

Parameters:

Name	Type	Description	Default
photos	PhotoEntityTyping	Фотографии в виде ссылки/пути до файла/сырых байтов/ IO-хранилища/Path-like объекта	()
peer_id	int	ID диалога или беседы, куда загружаются фотографии. Если не передавать, то фотографии загрузятся в скрытый альбом. Рекомендуется исключительно для тестирования, т.к. такой альбом имеет лимиты	0

Returns: Список врапперов загруженных фотографий, который можно напрямую передать в поле attachment при отправке сообщения

Source code in vkflow\api.py

async def upload_photos_to_message(self, *photos: PhotoEntityTyping, peer_id: int = 0) -> list[Photo]:
    """
    Загружает фотографию в сообщения

    Arguments:
        photos: Фотографии в виде ссылки/пути до файла/сырых байтов/
            IO-хранилища/Path-like объекта
        peer_id: ID диалога или беседы, куда загружаются фотографии. Если
            не передавать, то фотографии загрузятся в скрытый альбом. Рекомендуется
            исключительно для тестирования, т.к. такой альбом имеет лимиты
    Returns:
        Список врапперов загруженных фотографий, который можно напрямую
        передать в поле `attachment` при отправке сообщения
    """
    photo_bytes_coroutines = [self._fetch_file_entity(photo) for photo in photos]

    photo_bytes = await asyncio.gather(*photo_bytes_coroutines)

    async def upload_batch(batch: list[bytes]) -> list[Photo]:
        data_storage = aiohttp.FormData()

        for ind, photo in enumerate(batch):
            data_storage.add_field(
                f"file{ind}",
                photo,
                content_type="multipart/form-data",
                filename="a.png",
            )

        uploading_info = await self.method("photos.get_messages_upload_server", peer_id=peer_id)

        async with self.requests_session.post(
            uploading_info["upload_url"], data=data_storage
        ) as response:
            response = await self.parse_json_body(response, content_type=None)

        try:
            uploaded_photos = await self.method("photos.save_messages_photo", **response)
        except APIError[error_codes.CODE_1_UNKNOWN]:
            logger.exception(
                "Не удалось загрузить фотографии в беседу сообщества. "
                "VK пока не позволяет загружать фотографии в беседу сообщества"
            )

            return []
        else:
            return [Photo(uploaded_photo) for uploaded_photo in uploaded_photos]

    batches = [photo_bytes[i : i + 5] for i in range(0, len(photo_bytes), 5)]

    upload_coroutines = [upload_batch(batch) for batch in batches]
    results = await asyncio.gather(*upload_coroutines)

    result_photos = []

    for batch_result in results:
        result_photos.extend(batch_result)

    return result_photos

upload_doc_to_message async

upload_doc_to_message(
    content,
    filename,
    *,
    tags=None,
    return_tags=None,
    type="doc",
    peer_id=0
)

Загружает документ для отправки в сообщение

Parameters:

Name	Type	Description	Default
content	str | bytes	Содержимое документа. Документ может быть как текстовым, так и содержать сырые байты	required
filename	str	Имя файла	required
tags	str | None	Теги для файла, используемые при поиске	None
return_tags	bool | None	Возвращать переданные теги при запросе	None
type	Literal['doc', 'audio_message', 'graffiti']	Тип документа: файл/голосовое сообщение/граффити	'doc'
peer_id	int	ID диалога или беседы, куда загружается документ	0

Returns:

Type	Description
Document	Враппер загруженного документа. Этот объект можно напрямую
Document	передать в поле attachment при отправке сообщения

Source code in vkflow\api.py

async def upload_doc_to_message(
    self,
    content: str | bytes,
    filename: str,
    *,
    tags: str | None = None,
    return_tags: bool | None = None,
    type: typing.Literal["doc", "audio_message", "graffiti"] = "doc",
    peer_id: int = 0,
) -> Document:
    """
    Загружает документ для отправки в сообщение

    Arguments:
        content: Содержимое документа. Документ может быть
            как текстовым, так и содержать сырые байты
        filename: Имя файла
        tags: Теги для файла, используемые при поиске
        return_tags: Возвращать переданные теги при запросе
        type: Тип документа: файл/голосовое сообщение/граффити
        peer_id: ID диалога или беседы, куда загружается документ

    Returns:
        Враппер загруженного документа. Этот объект можно напрямую
        передать в поле `attachment` при отправке сообщения
    """
    if "." not in filename:
        filename = f"{filename}.txt"

    data_storage = aiohttp.FormData()
    data_storage.add_field(
        "file",
        content,
        content_type="multipart/form-data",
        filename=filename,
    )

    uploading_info = await self.method("docs.get_messages_upload_server", peer_id=peer_id, type=type)

    async with self.requests_session.post(uploading_info["upload_url"], data=data_storage) as response:
        response = await self.parse_json_body(response, content_type=None)

    document = await self.method(
        "docs.save",
        **response,
        tags=tags,
        title=filename,
        return_tags=return_tags,
    )

    return Document(document[type])

upload_video_to_message async

upload_video_to_message(
    file,
    *,
    name=None,
    description=None,
    is_private=True,
    wallpost=False,
    link=None,
    group_id=None,
    album_id=None,
    privacy_view=None,
    privacy_comment=None,
    no_comments=False,
    repeat=False,
    compression=False
)

Загружает видео для отправки в сообщение

Parameters:

Name	Type	Description	Default
file	str | bytes | BinaryIO | PathLike	Видео файл в виде ссылки/пути до файла/сырых байтов/IO-хранилища	required
name	str | None	Название видео	None
description	str | None	Описание видео	None
is_private	bool	Является ли видео приватным	True
wallpost	bool	Опубликовать видео на стене после сохранения	False
link	str | None	URL для встраивания видео с внешнего сайта	None
group_id	int | None	ID сообщества (для сообществ)	None
album_id	int | None	ID альбома, в который нужно загрузить видео	None
privacy_view	list[str] | None	Настройки приватности для просмотра	None
privacy_comment	list[str] | None	Настройки приватности для комментирования	None
no_comments	bool	Отключить комментарии	False
repeat	bool	Зациклить воспроизведение видео	False
compression	bool	Сжать видео для мобильных устройств	False

Returns:

Type	Description
Video	Враппер загруженного видео. Этот объект можно напрямую
Video	передать в поле attachments при отправке сообщения

Source code in vkflow\api.py

async def upload_video_to_message(
    self,
    file: str | bytes | typing.BinaryIO | os.PathLike,
    *,
    name: str | None = None,
    description: str | None = None,
    is_private: bool = True,
    wallpost: bool = False,
    link: str | None = None,
    group_id: int | None = None,
    album_id: int | None = None,
    privacy_view: list[str] | None = None,
    privacy_comment: list[str] | None = None,
    no_comments: bool = False,
    repeat: bool = False,
    compression: bool = False,
) -> VideoWrapper:
    """
    Загружает видео для отправки в сообщение

    Arguments:
        file: Видео файл в виде ссылки/пути до файла/сырых байтов/IO-хранилища
        name: Название видео
        description: Описание видео
        is_private: Является ли видео приватным
        wallpost: Опубликовать видео на стене после сохранения
        link: URL для встраивания видео с внешнего сайта
        group_id: ID сообщества (для сообществ)
        album_id: ID альбома, в который нужно загрузить видео
        privacy_view: Настройки приватности для просмотра
        privacy_comment: Настройки приватности для комментирования
        no_comments: Отключить комментарии
        repeat: Зациклить воспроизведение видео
        compression: Сжать видео для мобильных устройств

    Returns:
        Враппер загруженного видео. Этот объект можно напрямую
        передать в поле `attachments` при отправке сообщения
    """
    file_bytes = await self._fetch_file_entity(file)

    data_storage = aiohttp.FormData()
    data_storage.add_field(
        "video_file",
        file_bytes,
        content_type="multipart/form-data",
        filename=name or "video.mp4",
    )

    save_params = {
        "name": name,
        "description": description,
        "is_private": is_private,
        "wallpost": wallpost,
        "link": link,
        "group_id": group_id,
        "album_id": album_id,
        "privacy_view": privacy_view,
        "privacy_comment": privacy_comment,
        "no_comments": no_comments,
        "repeat": repeat,
        "compression": compression,
    }

    save_params = {k: v for k, v in save_params.items() if v is not None}
    uploading_info = await self.method("video.save", **save_params)

    async with self.requests_session.post(uploading_info["upload_url"], data=data_storage) as response:
        await self.parse_json_body(response, content_type=None)

    return VideoWrapper(
        {
            "id": uploading_info.get("video_id"),
            "owner_id": uploading_info.get("owner_id"),
            "title": name or "",
            "description": description or "",
        }
    )

TokenOwner

TokenOwner

Bases: Enum

Тип владельца токена: пользователь/группа/не определено