VKBottle

https://vkbottle.readthedocs.io/ru/latest/
Admin
VKBottle is an asynchronous Python framework designed for building bots and interacting with the...
Tokens:40,444
Snippets:366
Trust Score:9.6
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### Create AiohttpClient and call request_text

Source: https://vkbottle.readthedocs.io/ru/latest/low-level/http-client

Shows how to instantiate AiohttpClient and send a text request. The comment '# ...' indicates omitted setup code. The example uses the default session and calls request_text with a URL.

```python
from vkbottle.http import AiohttpClient

# ...
http_client = AiohttpClient()
await http_client.request_text("https://google.com")

```

--------------------------------

### Import aiohttp for async HTTP requests

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/faq

Example import of the aiohttp library, necessary for making async HTTP requests.

```python
import aiohttp

```

--------------------------------

### bot.py: Run the bot

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/code-separation

Starts the bot's polling loop.

```python
bot.run_forever()

```

--------------------------------

### Using Formatter and Format with message.answer()

Source: https://vkbottle.readthedocs.io/ru/latest/tools/formatting

Examples of using Formatter and Format functions with the BaseMessageMin.answer() method.

```python
await message.answer(Formatter("Hello, {:bold}!").format("World"))
await message.answer(underline(bold("Hello") + ", " + italic("World!")))

```

--------------------------------

### config.py: Initialize API, labeler, and state dispenser

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/code-separation

Creates the global API instance, a BotLabeler, and a BuiltinStateDispenser. The state dispenser is declared for clarity but not used in this example; import it from config.py and pass it to Bot in bot.py if needed.

```python
from vkbottle import API, BuiltinStateDispenser
from vkbottle.bot import BotLabeler


api = API("token")
labeler = BotLabeler()
state_dispenser = BuiltinStateDispenser()

```

--------------------------------

### Using formatting functions bold, italic, url

Source: https://vkbottle.readthedocs.io/ru/latest/tools/formatting

Examples of using the formatting functions bold, italic, and url to create formatted strings.

```python
bold("Hello, ") + italic("World!")  # Hello, World! ('Hello, ' is bold, 'World!' is italic)
"Hello, " + italic("World!")  # Hello, World! ('World!' is italic)
bold("Hello") + ", " + italic("World!")  # Hello, World! ('Hello' is bold, 'World!' is italic)
bold("vkbottle documentation:") + " " + url(italic("click me"), href="vkbottle.readthedocs.io/ru/latest")  # vkbottle documentation: click me ('click me' has an url)

```

--------------------------------

### bot.py: Imports for bot setup

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/code-separation

Imports Bot, the global API, state dispenser, labeler, and the chat and admin labelers. Because handlers/__init__.py exists, the labelers are already imported and no direct import from echo.py is needed.

```python
from vkbottle import Bot
from config import api, state_dispenser, labeler
from handlers import chat_labeler, admin_labeler

```

--------------------------------

### Async function to GET httpbin.org

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/faq

An async function that makes a GET request to httpbin.org using aiohttp, checking the status and returning the JSON response.

```python
async def send_httpbin_get() -> dict[str, Any]:
    async with aiohttp.ClientSession() as session:
        async with session.get('http://httpbin.org/get') as r:
            if r.status == 200:
                js = await r.json()
                return js

```

--------------------------------

### Add a task with LoopWrapper.add_task()

Source: https://vkbottle.readthedocs.io/ru/latest/tools/loop-wrapper

Creates a LoopWrapper instanceainer sharing a startup task and runs the event loop. The wrapper stores the task and executes it when the loop starts via run_forever. Uses add_task() to register the coroutine.

```python
from vkbottle import LoopWrapper

async def my_task():
    ...

lw = LoopWrapper()
lw.add_task(my_task())
lw.run()

```

--------------------------------

### Accessing API via Message.ctx_api

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/faq

Shows how to access the API through a Message object using message.ctx_api. The example bot responds to the text 'админы' and retrieves admin info.

```python
from vkbottle.bot import Bot, Message

bot = Bot(token="token")
admin_ids = [1, 100]


@bot.on.message(text="админы")
async def who_admins(message: Message):
    admins = [f"{adm.first_name} {adm.last_name}" for adm in (await message.ctx_api.users.get(admin_ids))]
    print((await message.ctx_api.request("users.get", {"user_ids": admin_ids})))
    await message.answer(f"Администраторы этого бота: {' '.join(admins)}")


bot.run_forever()

```

--------------------------------

### Register startup and shutdown coroutines with LoopWrapper

Source: https://vkbottle.readthedocs.io/ru/latest/tools/loop-wrapper

Appends coroutines to LoopWrapper.on_startup and LoopWrapper.on_shutdown. on_startup coroutines run before the event loop starts; on_shutdown coroutines run when it stops. The lists accept coroutine objects directly.

```python
from vkbottle import LoopWrapper

async def startup_task():
    print("This is startup")

async def shutdown_task():
    print("This is shutdown")

lw = LoopWrapper()
lw.on_startup.append(startup_task())
lw.on_shutdown.append(shutdown_task())
lw.run()

```

--------------------------------

### PayloadMapRule mapping example

Source: https://vkbottle.readthedocs.io/ru/latest/high-level/builtin-rules

Used to create a payload map that associates expected values and types with the message payload.

```json
{"name": "boba", "data": {"action": "do", "number": 12}}
```

```python
PayloadMapRule({"name": str, "data": {"action": str, "number": int}})
```

--------------------------------

### Bad: await at module level

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/faq

Incorrect example: calling an async function with await at module level raises SyntaxError: 'await' outside function.

```python
# bad
async def main():
    print("main")


await main()  # SyntaxError: 'await' outside function

```

--------------------------------

### Bad: await outside async function

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/faq

Incorrect example: using await outside an async function raises SyntaxError: 'await' outside async function.

```python
# bad
async def foo() -> str: return "foo"


def main():
    print(await foo()) # SyntaxError: 'await' outside async function


main()

```

--------------------------------

### API Initialization and Usage

Source: https://vkbottle.readthedocs.io/ru/latest/low-level/api

Initializes the VKBottle API client with a token and demonstrates how to call a method, using snake_case naming.

```APIDOC
## API

### Description
Create an API client instance and call VK API methods. Methods are named in snake_case, e.g., `messages.getById` becomes `api.messages.get_by_id`.

### Usage Example
```python
from vkbottle import API

api = API(token="token")
await api.messages.send(peer_id=1, message="Привет Павел Дуров!")
```

### Parameters
- **token** - community/user token or token generator
- **ignore_error** - ignore VK API errors
- **http_client** - client for requests (documentation)
- **request_rescheduler** - request rescheduler (documentation)
```

--------------------------------

### Good: await inside async function

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/faq

Correct example: await is used inside an async function, and the program is started with asyncio.run().

```python
# good
import asyncio

async def foo() -> str: return "foo"


async def main():
    print(await foo())


asyncio.run(main())

```

--------------------------------

### bot.py: Load labelers and run bot

Source: https://vkbottle.readthedocs.io/ru/latest/high-level/bot/labeler

Creates a Bot instance, loads each custom labeler from the routes module, and runs the bot. Requires a valid token and the routes module with a labelers list.

```python
from vkbottle import Bot
from routes import labelers

bot = Bot("token")

for custom_labeler in labelers:
    bot.labeler.load(custom_labeler)

bot.run_forever()
```

--------------------------------

### bot.py: Create Bot instance

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/code-separation

Creates the Bot instance, passing the API, the global labeler, and the state dispenser.

```python
bot = Bot(
    api=api,
    labeler=labeler,
    state_dispenser=state_dispenser,
)

```

--------------------------------

### routes/__init__.py: Aggregate labelers

Source: https://vkbottle.readthedocs.io/ru/latest/high-level/bot/labeler

Imports labelers from greetings and goodbyes modules and exposes them as a list for the main bot to load.

```python
from . import greetings, goodbyes

labelers = [greetings.bl, goodbyes.bl]
```

--------------------------------

### Formatter.format with positional and keyword arguments

Source: https://vkbottle.readthedocs.io/ru/latest/tools/formatting

Shows how to use the format method with *args and **kwargs to replace {} placeholders.

```python
Formatter("Hi, {}.").format("Alex")  # Hi, Alex.
Formatter("Hi, {name}.").format(name="Maria")  # Hi, Maria.

```

--------------------------------

### Good: Using AiohttpClient

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/faq

Shows the recommended approach using vkbottle's AiohttpClient for non-blocking HTTP requests.

```python
# good, recommended
# see https://vkbottle.rtfd.io/ru/latest/low-level/http-client/
from vkbottle.http import AiohttpClient


async def send_httpbin_get() -> dict[str, Any]:
    http_client = AiohttpClient()
    return await http_client.request_json('http://httpbin.org/get')
```

--------------------------------

### __init__.py: Import labelers and define __all__

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/code-separation

Imports all labelers from the handlers package and defines __all__. The comment notes that handlers are registered in import order when using the global labeler.

```python
from .chat import chat_labeler
from .admin import admin_labeler
from .ping import labeler
# Если использовать глобальный лейблер, то все хендлеры будут зарегистрированы в том же порядке, в котором они были импортированы

__all__ = ("admin_labeler", "chat_labeler", "labeler")

```

--------------------------------

### Await an async function to get its result

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/faq

Shows that calling an async function without await returns a coroutine object, while await executes it and returns the result. Run with asyncio.run().

```python
import asyncio


async def get_banana() -> str: return "banana"


async def main():
    print(get_banana())  # <coroutine object get_banana at 0x...>
    print(await get_banana())  # banana


asyncio.run(main())

```

--------------------------------

### Initialize API and send message

Source: https://vkbottle.readthedocs.io/ru/latest/low-level/api

Initialize the VK API client with a token and send a message. All methods use snake_case, e.g., messages.getById becomes api.messages.get_by_id.

```python
from vkbottle import API

# ...
api = API(token="token")
await api.messages.send(peer_id=1, message="Привет Павел Дуров!")
```

--------------------------------

### Get token with custom scope using UserPermission

Source: https://vkbottle.readthedocs.io/ru/latest/tools/auth

Obtain a token with a custom permission scope by passing a list of UserPermission values to the scope parameter of get_token. The default scope can be overridden this way.

```python
from vkbottle import UserAuth, UserPermission

login = "89012345678"
password = "qwerty123"

async def main():
    token = await UserAuth().get_token(login, password, scope=[UserPermission.photos, UserPermission.video])

```

--------------------------------

### Get user token with UserAuth

Source: https://vkbottle.readthedocs.io/ru/latest/tools/auth

Basic usage of UserAuth to obtain a user token by login and password. By default, the mobile VK application is used for login and a default set of permissions is requested.

```python
from vkbottle import UserAuth

login = "89012345678"
password = "qwerty123"

async def main():
    token = await UserAuth().get_token(login, password)

```

--------------------------------

### run_multibot function signature

Source: https://vkbottle.readthedocs.io/ru/latest/high-level/bot/multibot

Starts a multibot by adding run_polling tasks to the event loop. The bot parameter does not need a token for API creation; apis can be any iterable, and polling_type can be customized if you have a custom polling implementation.

```python
run_multibot(bot, apis, polling_type=BotPolling)
```

--------------------------------

### Handle captcha and 2FA errors when getting token

Source: https://vkbottle.readthedocs.io/ru/latest/tools/auth

Handles captcha, SMS 2FA, and 2FA app validation during token retrieval. Uses CaptchaError, APIAuthError, and AuthError exceptions, and calls validate_phone for SMS validation.

```python
from __future__ import annotations

import asyncio
from typing import TYPE_CHECKING, cast
from urllib.parse import parse_qs, urlparse

from vkbottle import APIAuthError, AuthError, CaptchaError, UserAuth

if TYPE_CHECKING:
    from typing import Optional, Union

login = "89012345678"
password = "qwerty123"

async def main():
    user_auth = UserAuth()

    auth_code: bool | str = True
    captcha_sid: str | None = None
    captcha_key: str | None = None

    while True:
        try:
            token = await user_auth.get_token(
                login,
                password,
                auth_code=auth_code,
                captcha_sid=captcha_sid,
                captcha_key=captcha_key
            )
            break
        except CaptchaError as error:
            captcha_sid = error.captcha_sid
            captcha_key = input(f"Open url '{error.captcha_img}' in browser and enter captcha code: ")
        except APIAuthError as error:
            if error.validation_type == "2fa_sms":
                validation_sid = cast(str, error.validation_sid)
                await user_auth.validate_phone(validation_sid)
                auth_code = input("Enter code from SMS: ")
                continue

            if error.validation_type == "2fa_app":
                print(f"Enter code from 2FA app here: {error.redirect_uri}")
                oauth_url = input("Enter OAuth URL from VK after signin in: ")
                parsed_url = urlparse(url=oauth_url)
                token = parse_qs(parsed_url.fragment)["access_token"][0]
                break

            print(error.error_description)
            return
        except AuthError as error:
            print(error.error_description)
            return

    print(f"Your token: {token}")

```

--------------------------------

### Использование шортката command для правила

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/rules

Использует шорткат command в декораторе @bot.on.message. Параметры правила передаются в кортеже: команда и количество аргументов. Хендлер получает аргументы как кортеж.

```python
@bot.on.message(command=("say", 1))
async def say_handler(message: Message, args: tuple[str]):
    await message.answer(f"<<{args[0]}>>")

```

--------------------------------

### Define states and handle messages with StateRule

Source: https://vkbottle.readthedocs.io/ru/latest/high-level/handling/state-dispenser

Shows how to define a state group with BaseStateGroup, set a state using bot.state_dispenser.set, and handle messages with a state rule. The handler for state SuperStates.AWKWARD_STATE responds with 'oi awkward', and the /die command sets that state and returns 'ok'.

```python
from vkbottle import BaseStateGroup
from vkbottle.bot import Message, Bot

bot = Bot("t")

class SuperStates(BaseStateGroup):
    AWKWARD_STATE = "awkward"
    CONFIDENT_STATE = "confident"
    TERRIFYING_STATE = "terrifying"

@bot.on.message(state=SuperStates.AWKWARD_STATE)  # StateRule(SuperStates.AWKWARD_STATE)
async def awkward_handler(message: Message):
    await message.answer("oi awkward")

@bot.on.message(lev="/die")
async def die_handler(message: Message):
    await bot.state_dispenser.set(message.peer_id, SuperStates.AWKWARD_STATE)
    return "ok"

bot.run_forever()
```

--------------------------------

### Инициализация бота с callback

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/callback-bot

Инициализация бота с токеном и callback-объектом.

```python
from vkbottle import Bot
bot = Bot(token=TOKEN, callback=callback)
```

--------------------------------

### Custom Response Validator with ABCResponseValidator

Source: https://vkbottle.readthedocs.io/ru/latest/low-level/api/response-validator

Example of creating a custom response validator that handles VK API error code 6 (too many requests) by rescheduling the request. It must be inserted after JSONResponseValidator but before VKAPIErrorResponseValidator to avoid handling the error prematurely.

```python
from typing import TYPE_CHECKING, Any, NoReturn, Union
from vkbottle import API, ABCResponseValidator
from .request_rescheduler import SomeRequestRescheduler

if TYPE_CHECKING:
    from vkbottle.api import ABCAPI, API

class SomeResponseValidator(ABCResponseValidator):
    async def validate(
        self,
        method: str,
        data: dict[str, Any]:
        response: Any,
        ctx_api: "ABCAPI | API",
    ) -> Any | NoReturn:
        if "error" not in response:
            return response
        if ctx_api.ignore_errors:
            return None
        if response["error"]["error_code"] != 6:
            return response
        # Обрабатываем ошибку 6, которая возникает при слишком частых запросах
        return await SomeRequestRescheduler().reschedule(
            ctx_api, method, data, response
        )

api = API("token")
# Этот валидатор должен идти после JSONResponseValidator,
# но раньше VKAPIErrorResponseValidator, чтобы он не обработал эту ошибку
api.response_validators.insert(1, SomeResponseValidator())
```

--------------------------------

### Инициализация параметров CallbackAPI

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/callback-bot

Инициализация параметров для настройки CallbackAPI: токен сообщества, URL, заголовок и опциональный секретный ключ. Значения берутся из переменных окружения.

```python
import os
TOKEN = os.getenv("VK_TOKEN") # ключ сообщества
url = os.getenv("VK_URL") # url = "http://example.com/whateveryouwnant"
title = os.getenv("VK_TITLE") # title = "server"
secret_key = os.getenv("VK_SECRET_KEY") # опционально
```

--------------------------------

### Custom Request Rescheduler Implementation

Source: https://vkbottle.readthedocs.io/ru/latest/low-level/api/request-rescheduler

Example of a custom rescheduler that implements the `reschedule` method. It sleeps for a configurable delay and then retries the request using `ctx_api.request`. The method signature includes `ctx_api`, `method`, `data`, and `recent_response` as required by the `ABCRequestRescheduler` interface.

```python
import asyncio
from typing import TYPE_CHECKING, Any, Union

from vkbottle import ABCRequestRescheduler

if TYPE_CHECKING:
    from vkbottle.api import ABCAPI, API

class SomeRequestRescheduler(ABCRequestRescheduler):
    def __init__(self, delay: int = 1):
        self.delay = delay

    async def reschedule(
        self,
        ctx_api: "ABCAPI | API",
        method: str,
        data: dict[str, Any]:
        recent_response: Any,
    ) -> dict[str, Any]:
        await asyncio.sleep(self.delay)
        return await ctx_api.request(method, data)
```

--------------------------------

### bot.py: Load labelers into global labeler

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/code-separation

Loads the chat and admin labelers into the global labeler so their handlers are registered.

```python
labeler.load(chat_labeler)
labeler.load(admin_labeler)

```

--------------------------------

### Convert Python function to VKScript with @vkscript decorator

Source: https://vkbottle.readthedocs.io/ru/latest/tools/vkscript

Use the @vkscript decorator from vkbottle to wrap a function that uses the api object. The function's logic is converted to VKScript code, which can be executed via api.execute(). The example shows sending messages to multiple users and returning message IDs.

```python
from vkbottle import vkscript

@vkscript
def my_execute(api, user_ids=()):
    message_ids = []
    for user_id in user_ids:
        user = api.users.get(user_ids=user_id)[0]
        message_ids.append(api.messages.send(message=f"{user.first_name}, спасибо что зашел на чай", random_id=0, peer_id=user_id))
    return message_ids

# далее вызывая эту функцию вы можете использовать ее ответ в качестве кода для execute

await api.execute(code=my_execute(user_ids=[10, 11, 12]))
```

--------------------------------

### Запуск обернутой функции через asyncio.run

Source: https://vkbottle.readthedocs.io/ru/latest/low-level/exception_handling/error-handler

Запускает асинхронную функцию main с помощью asyncio.run, что приводит к обработке исключения LookupError и выводу сообщения.

```python
import asyncio

asyncio.run(main())
```

--------------------------------

### Import Formatter from vkbottle.tools.formatting

Source: https://vkbottle.readthedocs.io/ru/latest/tools/formatting

Import statement for the Formatter class used for formatted text.

```python
from vkbottle.tools.formatting import Formatter

```

--------------------------------

### Basic bot message handler with vkbottle Bot

Source: https://vkbottle.readthedocs.io/ru/latest/high-level/handling/handler

Creates a vkbottle bot with a token and registers a blocking handler for all messages using @bot.on.message(). The handler replies with a greeting. Run with bot.run_forever().

```python
from vkbottle.bot import Bot, Message

bot = Bot("token")

@bot.on.message()
async def any_message(message: Message):
    await message.answer("Привет я бот")

bot.run_forever()
```

--------------------------------

### admin.py: Admin labeler with FromPeerRule and halt command

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/code-separation

Defines an admin labeler restricted to a specific user ID via FromPeerRule(1). The handler for the "halt" command exits the process.

```python
from vkbottle.bot import BotLabeler, Message, rules

admin_labeler = BotLabeler()
admin_labeler.auto_rules = [rules.FromPeerRule(1)] # Допустим, вы являетесь Павлом Дуровым

@admin_labeler.message(command="halt")
async def halt(_):
    exit(0)

```

--------------------------------

### Import Format from vkbottle.tools.formatting

Source: https://vkbottle.readthedocs.io/ru/latest/tools/formatting

Import statement for the Format class used to represent formatting types.

```python
from vkbottle.tools.formatting import Format

```

--------------------------------

### Set vkbottle logging level with logging

Source: https://vkbottle.readthedocs.io/ru/latest/modules

Set the logging level for the 'vkbottle' logger to INFO using the standard logging library.

```python
import logging
logging.getLogger("vkbottle").setLevel(logging.INFO)

```

--------------------------------

### Использование CommandRule из vkbottle.dispatch.rules.base

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/rules

Импортирует CommandRule и использует его в декораторе @bot.on.message. Правило принимает команду, префиксы и количество аргументов. Хендлер получает аргументы как кортеж.

```python
from vkbottle.dispatch.rules.base import CommandRule

@bot.on.message(CommandRule("say", ["!", "/"], 1))
async def say_handler(message: Message, args: tuple[str]):
    await message.answer(f"<<{args[0]}>>")

```

--------------------------------

### Formatter.format_map with Mapping argument

Source: https://vkbottle.readthedocs.io/ru/latest/tools/formatting

Demonstrates the format_map method which accepts a Mapping argument.

```python
Formatter("My bestie is {bestie:underline}").format_map({"bestie": "telegrinder")  # My bestie is telegrinder

```

--------------------------------

### Combine formatting types with '+'

Source: https://vkbottle.readthedocs.io/ru/latest/tools/formatting

Shows how to combine multiple formatting types using the '+' symbol.

```python
Formatter("Very cool {:bold+italic} ^_^").format("bold-italic message")  # Very cool bold-italic message ^_^

```

--------------------------------

### Set state with payload and access it

Source: https://vkbottle.readthedocs.io/ru/latest/high-level/handling/state-dispenser

Shows how to pass a payload to state_dispenser.set and later access it via message.state_peer.payload. The payload key 'something' is set to 1 and then printed.

```python
# ... In handler:
await bot.state_dispenser.set(message.peer_id, SuperStates.TERRIFYING_STATE, something=1)
# ... With state:
print(message.state_peer.payload["something"])  # 1
```

--------------------------------

### Кастомизация MyRule через __init__

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/rules

Добавляет параметр lt (по умолчанию 100) в конструктор правила. Метод check сравнивает длину текста с self.lt.

```python
# Новый вид правила
class MyRule(ABCRule[Message]):
    def __init__(self, lt: int = 100):
        self.lt = lt

    async def check(self, event: Message) -> bool:
        return len(event.text) < self.lt

```

--------------------------------

### Инициализация Bot токеном или API

Source: https://vkbottle.readthedocs.io/ru/latest/tutorial/first-bot

Два способа инициализации бота: токеном (создаётся API с генератором токена) или готовым API.

```python
# Инициализация токеном
bot = Bot(token="token")
# (или) Инициализация с апи
bot = Bot(api=api)

```
Tokens:
Raw