Утилиты

DotDict

DotDict

DotDict(data=None, **kwargs)

Bases: dict

Словарь с поддержкой доступа к ключам через атрибуты (точку).

Позволяет обращаться к ключам как через obj["key"], так и через obj.key. При этом isinstance(obj, dict) возвращает True.

Вложенные словари автоматически оборачиваются в DotDict, а списки с вложенными словарями также обрабатываются рекурсивно.

Examples:

data = DotDict({"user": {"name": "Alice", "age": 25}})
data.user.name   # 'Alice'
data["user"]["age"]  # 25
data.user.city = "Moscow"
data.user.city   # 'Moscow'
isinstance(data, dict)  # True

Source code in vkflow\utils\dotdict.py

def __init__(self, data: dict[str, typing.Any] | None = None, **kwargs: typing.Any) -> None:
    if data is None:
        data = {}
    data.update(kwargs)

    for key, value in data.items():
        data[key] = self._wrap(value)

    super().__init__(data)

dict_keys

dict_keys()

Аналог dict.keys() -используется когда в словаре есть ключ "keys".

Source code in vkflow\utils\dotdict.py

def dict_keys(self):
    """Аналог dict.keys() -используется когда в словаре есть ключ "keys"."""
    return dict.keys(self)

dict_values

dict_values()

Аналог dict.values() -используется когда в словаре есть ключ "values".

Source code in vkflow\utils\dotdict.py

def dict_values(self):
    """Аналог dict.values() -используется когда в словаре есть ключ "values"."""
    return dict.values(self)

dict_items

dict_items()

Аналог dict.items() -используется когда в словаре есть ключ "items".

Source code in vkflow\utils\dotdict.py

def dict_items(self):
    """Аналог dict.items() -используется когда в словаре есть ключ "items"."""
    return dict.items(self)

Хелперы

helpers

peer

peer(chat_id=0)

Конвертирует chat_id в peer_id (chat_id + 2_000_000_000).

Source code in vkflow\utils\helpers.py

def peer(chat_id: int = 0) -> int:
    """Конвертирует chat_id в peer_id (chat_id + 2_000_000_000)."""
    return 2_000_000_000 + chat_id

random_id

random_id(side=2 ** 31 - 1)

Случайное число в диапазоне +-side. Используется для API метода messages.send.

Source code in vkflow\utils\helpers.py

def random_id(side: int = 2**31 - 1) -> int:
    """
    Случайное число в диапазоне +-`side`.
    Используется для API метода `messages.send`.
    """
    return random.randint(-side, +side)

Медиа

media

download_file async

download_file(url, *, session=None, **kwargs)

Скачивание файлов по их прямой ссылке.

Source code in vkflow\utils\media.py

async def download_file(
    url: str,
    *,
    session: aiohttp.ClientSession | None = None,
    **kwargs,
) -> bytes:
    """
    Скачивание файлов по их прямой ссылке.
    """
    if session is not None:
        async with session.get(url, **kwargs) as response:
            return await response.read()

    async with (
        aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(ssl=ssl.create_default_context()),
            skip_auto_headers={"User-Agent"},
            raise_for_status=True,
            json_serialize=json_parser_policy.dumps,
        ) as new_session,
        new_session.get(url, **kwargs) as response,
    ):
        return await response.read()

get_user_registration_date async

get_user_registration_date(id_, *, session=None)

Получает дату регистрации пользователя ВКонтакте через публичный FOAF endpoint.

Source code in vkflow\utils\media.py

async def get_user_registration_date(
    id_: int, *, session: aiohttp.ClientSession | None = None
) -> datetime.datetime:
    """
    Получает дату регистрации пользователя ВКонтакте
    через публичный FOAF endpoint.
    """
    request_session = session or aiohttp.ClientSession(
        connector=aiohttp.TCPConnector(ssl=False),
        skip_auto_headers={"User-Agent"},
        raise_for_status=True,
        json_serialize=json_parser_policy.dumps,
    )
    async with (
        request_session,
        request_session.get("https://vk.ru/foaf.php", params={"id": id_}) as response,
    ):
        user_info = await response.text()
        registration_date = _registration_date_regex.search(user_info)
        if registration_date is None:
            raise ValueError(f"No such user with id `{id_}`")
        registration_date = registration_date.group("date")
        return datetime.datetime.fromisoformat(registration_date)

JSON-парсеры

json_parsers

Имплементации разных JSON парсеров

json_parser_policy module-attribute

json_parser_policy = _select_parser()

BuiltinJsonParser

Bases: BaseJSONParser

OrjsonParser

Bases: BaseJSONParser

UjsonParser

Bases: BaseJSONParser

Форматирование

formatting

FormatSegment dataclass

FormatSegment(type, offset, length, data=None)

MarkdownParser

format_message

format_message(text)

Source code in vkflow\formatting.py

def format_message(text: str | None) -> tuple[str | None, dict[str, typing.Any] | None]:
    if text is None:
        return None, None

    plain_text, format_data = MarkdownParser.parse(text)

    if not format_data["items"]:
        return plain_text, None

    return plain_text, format_data