BaseConnection

NamedBytesIO(buffer=b'', *, name=None)

Bases: BytesIO

BytesIO с поддержкой атрибута .name для единообразия с файловыми объектами.

Source code in maxapi/connection/base.py

def __init__(
    self, buffer: bytes = b"", *, name: str | None = None
) -> None:
    super().__init__(buffer)
    self.name = name  # Соответствует протоколу typing.BinaryIO

BaseConnection()

Bases: BotMixin

Базовый класс для всех методов API.

Содержит общую логику выполнения запроса (сериализация, отправка HTTP-запроса, обработка ответа).

Инициализация BaseConnection.

Атрибуты

bot: Экземпляр бота. session: aiohttp-сессия. after_input_media_delay: Задержка после ввода медиа.

Source code in maxapi/connection/base.py

def __init__(self) -> None:
    """
    Инициализация BaseConnection.

    Атрибуты:
        bot: Экземпляр бота.
        session: aiohttp-сессия.
        after_input_media_delay: Задержка после ввода медиа.
    """

    self.bot: Bot | None = None
    self.session: ClientSession | None = None
    self.after_input_media_delay: float = self.AFTER_MEDIA_INPUT_DELAY
    self.api_url = self.API_URL

set_api_url(url)

Установка API URL для запросов

Parameters:

Name	Type	Description	Default
url	str	Новый API URL	required

Source code in maxapi/connection/base.py

def set_api_url(self, url: str) -> None:
    """
    Установка API URL для запросов

    Args:
        url: Новый API URL
    """

    self.api_url = url

request(method, path, model=None, *, is_return_raw=False, **kwargs) async

Выполняет HTTP-запрос к API с автоматическим retry при серверных ошибках.

При получении HTTP-статуса из списка retry_on_statuses (по умолчанию 502, 503, 504) запрос повторяется до max_retries раз с экспоненциальной задержкой.

Parameters:

Name	Type	Description	Default
method	HTTPMethod	HTTP-метод (GET, POST и т.д.).	required
path	ApiPath | str	Путь до конечной точки.	required
model	BaseModel | Any	Pydantic-модель для десериализации ответа, если is_return_raw=False.	None
is_return_raw	bool	Если True — вернуть сырой ответ, иначе — результат десериализации.	False
**kwargs	Any	Дополнительные параметры (query, headers, json).	{}

Returns:

Type	Description
Any | BaseModel	model | dict | Error: Объект модели, dict или ошибка.

Raises:

Type	Description
RuntimeError	Если бот не инициализирован.
MaxConnection	Ошибка соединения.
InvalidToken	Ошибка авторизации (401).
MaxApiError	Ошибка API (после исчерпания retry).

Source code in maxapi/connection/base.py

async def request(
    self,
    method: HTTPMethod,
    path: ApiPath | str,
    model: BaseModel | Any = None,
    *,
    is_return_raw: bool = False,
    **kwargs: Any,
) -> Any | BaseModel:
    """
    Выполняет HTTP-запрос к API с автоматическим retry
    при серверных ошибках.

    При получении HTTP-статуса из списка ``retry_on_statuses``
    (по умолчанию 502, 503, 504) запрос повторяется до
    ``max_retries`` раз с экспоненциальной задержкой.

    Args:
        method: HTTP-метод (GET, POST и т.д.).
        path: Путь до конечной точки.
        model: Pydantic-модель для
            десериализации ответа, если is_return_raw=False.
        is_return_raw: Если True — вернуть сырой
            ответ, иначе — результат десериализации.
        **kwargs: Дополнительные параметры (query, headers, json).

    Returns:
        model | dict | Error: Объект модели, dict или ошибка.

    Raises:
        RuntimeError: Если бот не инициализирован.
        MaxConnection: Ошибка соединения.
        InvalidToken: Ошибка авторизации (401).
        MaxApiError: Ошибка API (после исчерпания retry).
    """

    bot = self._ensure_bot()
    conn = bot.default_connection
    retry_statuses = conn.retry_on_statuses

    url = path.value if isinstance(path, ApiPath) else path

    @backoff.on_exception(
        backoff.expo,
        (ClientConnectionError, _RetryableServerError),
        max_tries=conn.max_retries + 1,
        factor=conn.retry_backoff_factor,
        on_backoff=_on_backoff,
    )
    async def _do_request() -> Any:
        session = await bot.ensure_session()
        resp = await session.request(
            method=method.value,
            url=url,
            **kwargs,
        )

        if resp.status == 401:
            await session.close()
            raise InvalidToken("Неверный токен!")

        if resp.status in retry_statuses:
            await resp.read()
            raise _RetryableServerError(resp.status)

        return resp

    try:
        response = await _do_request()
    except ClientConnectionError as e:
        raise MaxConnection(f"Ошибка при отправке запроса: {e}") from e
    except _RetryableServerError as e:
        raise MaxApiError(code=e.status, raw={"error": str(e)}) from e

    if not response.ok:
        raw = await response.json()
        if bot.dispatcher:
            asyncio.create_task(
                bot.dispatcher.handle_raw_response(
                    UpdateType.RAW_API_RESPONSE, raw
                )
            )
        raise MaxApiError(code=response.status, raw=raw)

    raw = await response.json()

    if bot.dispatcher:
        asyncio.create_task(
            bot.dispatcher.handle_raw_response(
                UpdateType.RAW_API_RESPONSE, raw
            )
        )

    if is_return_raw:
        return raw

    model = model(**raw)  # type: ignore

    return bind_bot(model, bot)

upload_file(url, path, type) async

Загружает файл на сервер.

Parameters:

Name	Type	Description	Default
url	str	URL загрузки.	required
path	str	Путь к файлу.	required
type	UploadType	Тип файла.	required

Returns:

Name	Type	Description
str	str	Сырой .text() ответ от сервера.

Source code in maxapi/connection/base.py

async def upload_file(self, url: str, path: str, type: UploadType) -> str:
    """
    Загружает файл на сервер.

    Args:
        url: URL загрузки.
        path: Путь к файлу.
        type: Тип файла.

    Returns:
        str: Сырой .text() ответ от сервера.
    """

    async with aiofiles.open(path, "rb") as f:
        file_data = await f.read()

    path_object = Path(path)
    basename = path_object.name

    form = FormData(quote_fields=False)
    form.add_field(
        name="data",
        value=file_data,
        filename=basename,
        content_type=mimetypes.guess_type(path)[0] or f"{type.value}/*",
    )

    bot = self._ensure_bot()

    session = bot.session
    if session is not None and not session.closed:
        async with session.post(url=url, data=form) as response:
            return await response.text()
    else:
        async with (
            ClientSession(
                timeout=bot.default_connection.timeout,
                **connector_kwargs(bot.default_connection.kwargs),
            ) as temp_session,
            temp_session.post(url=url, data=form) as response,
        ):
            return await response.text()

upload_file_buffer(filename, url, buffer, type) async

Загружает файл из буфера.

Parameters:

Name	Type	Description	Default
filename	str	Имя файла.	required
url	str	URL загрузки.	required
buffer	bytes	Буфер данных.	required
type	UploadType	Тип файла.	required

Returns:

Name	Type	Description
str	str	Сырой .text() ответ от сервера.

Source code in maxapi/connection/base.py

async def upload_file_buffer(
    self, filename: str, url: str, buffer: bytes, type: UploadType
) -> str:
    """
    Загружает файл из буфера.

    Args:
        filename: Имя файла.
        url: URL загрузки.
        buffer: Буфер данных.
        type: Тип файла.

    Returns:
        str: Сырой .text() ответ от сервера.
    """

    try:
        matches = puremagic.magic_string(buffer[:4096])
        if matches:
            mime_type = matches[0].mime_type
            ext = mimetypes.guess_extension(mime_type) or ""
        else:
            mime_type = f"{type.value}/*"
            ext = ""
    except (OSError, ValueError, AttributeError):
        mime_type = f"{type.value}/*"
        ext = ""

    basename = f"{filename}{ext}"

    form = FormData(quote_fields=False)
    form.add_field(
        name="data",
        value=buffer,
        filename=basename,
        content_type=mime_type,
    )

    bot = self._ensure_bot()

    session = bot.session
    if session is not None and not session.closed:
        async with session.post(url=url, data=form) as response:
            return await response.text()
    else:
        async with (
            ClientSession(
                timeout=bot.default_connection.timeout,
                **connector_kwargs(bot.default_connection.kwargs),
            ) as temp_session,
            temp_session.post(url=url, data=form) as response,
        ):
            return await response.text()

download_file(url, destination, *, filename=None, chunk_size=DOWNLOAD_CHUNK_SIZE) async

Скачивает файл по URL и сохраняет на диск.

URL можно получить из payload вложения: - Изображение: attachment.payload.url - Видео: attachment.urls.mp4_720 (или другое разрешение) - Аудио/Файл: attachment.payload.url - Стикер: attachment.payload.url

Метод работает не через общий request(), поскольку ответом является бинарный поток, а не JSON.

Если файл существует, то возвращает новый свободный путь для сохранения

Windows style: - file_name.ext - file_name(2).ext - file_name(3).ext

Parameters:

Name	Type	Description	Default
url	str	URL файла для скачивания (из payload.url вложения).	required
destination	Path | str	Путь к директории для сохранения файла.	required
filename	Path | str | None	Имя файла для сохранения. Если не указано, то будет использовано имя, предоставляемое сервером или значение по умолчанию.	None
chunk_size	int	Размер чанка при потоковом чтении (по умолчанию 64 КБ).	DOWNLOAD_CHUNK_SIZE

Returns:

Name	Type	Description
Path	Path	Полный путь к скачанному файлу.

Raises:

Type	Description
DownloadFileError	при ошибке скачивания.
(FileExistsError, NotADirectoryError, PermissionError, OSError)	при ошибках файловой системы

Source code in maxapi/connection/base.py

async def download_file(
    self,
    url: str,
    destination: Path | str,
    *,
    filename: Path | str | None = None,
    chunk_size: int = DOWNLOAD_CHUNK_SIZE,
) -> Path:
    """
    Скачивает файл по URL и сохраняет на диск.

    URL можно получить из payload вложения:
    - Изображение: ``attachment.payload.url``
    - Видео: ``attachment.urls.mp4_720`` (или другое разрешение)
    - Аудио/Файл: ``attachment.payload.url``
    - Стикер: ``attachment.payload.url``

    Метод работает не через общий ``request()``, поскольку
    ответом является бинарный поток, а не JSON.

    Если файл существует, то возвращает новый свободный путь для сохранения

    Windows style:
    - file_name.ext
    - file_name(2).ext
    - file_name(3).ext

    Args:
        url: URL файла для скачивания (из payload.url вложения).
        destination: Путь к директории для сохранения файла.
        filename: Имя файла для сохранения. Если не указано,
            то будет использовано имя, предоставляемое сервером
            или значение по умолчанию.
        chunk_size: Размер чанка при потоковом чтении
            (по умолчанию 64 КБ).

    Returns:
        Path: Полный путь к скачанному файлу.

    Raises:
        DownloadFileError: при ошибке скачивания.
        FileExistsError, NotADirectoryError, PermissionError, OSError:
            при ошибках файловой системы
    """
    dest = Path(destination)
    final_path = None

    # Получаем ответ для определения имени файла из заголовков
    response = await self._fetch_response(url)

    try:
        await aiofiles.os.makedirs(dest, exist_ok=True)
    except (FileExistsError, NotADirectoryError, PermissionError, OSError):
        # Если передан файл вместо директории, путь ошибочен
        # или нет прав доступа
        response.release()
        raise

    try:
        if filename:
            # Выделяем только имя файла,
            # в случае если переменная содержит путь
            filename = Path(filename).name
        else:
            filename = self._capture_filename(response)

        final_path = self._check_file_exists(dest / filename)
        async with aiofiles.open(final_path, "wb") as f:
            async for chunk in self._fetch_content_stream(
                response, chunk_size=chunk_size
            ):
                await f.write(chunk)
    except Exception:
        # При любой ошибке удаляем частично записанный файл
        if final_path and final_path.exists():
            final_path.unlink()
        raise
    finally:
        response.release()

    return final_path

download_bytes_io(url, *, chunk_size=DOWNLOAD_CHUNK_SIZE) async

Скачивает файл по URL и возвращает file-like объект в памяти.

Внимание: весь файл загружается в оперативную память. Не используйте для файлов >100–200 МБ без контроля.

Parameters:

Name	Type	Description	Default
url	str	URL файла.	required
chunk_size	int	Размер чанка при потоковом чтении.	DOWNLOAD_CHUNK_SIZE

Returns:

Name	Type	Description
NamedBytesIO	NamedBytesIO	Содержимое файла с атрибутом .name.
	NamedBytesIO	Наследуется от io.BytesIO
	NamedBytesIO	Для zero-copy передачи используйте .getbuffer(),
	NamedBytesIO	для получения bytes — .read() или .getvalue().

Raises:

Type	Description
DownloadFileError	при ошибке скачивания.

Source code in maxapi/connection/base.py

async def download_bytes_io(
    self,
    url: str,
    *,
    chunk_size: int = DOWNLOAD_CHUNK_SIZE,
) -> NamedBytesIO:
    """
    Скачивает файл по URL и возвращает file-like объект в памяти.

    Внимание: весь файл загружается в оперативную память.
    Не используйте для файлов >100–200 МБ без контроля.

    Args:
        url: URL файла.
        chunk_size: Размер чанка при потоковом чтении.

    Returns:
        NamedBytesIO: Содержимое файла с атрибутом .name.
        Наследуется от io.BytesIO
        Для zero-copy передачи используйте .getbuffer(),
        для получения bytes — .read() или .getvalue().

    Raises:
        DownloadFileError: при ошибке скачивания.
    """
    bio = NamedBytesIO()

    response = await self._fetch_response(url)
    bio.name = self._capture_filename(response)

    async for chunk in self._fetch_content_stream(
        response,
        chunk_size=chunk_size,
    ):
        bio.write(chunk)

    bio.seek(0)  # обязательно переходим в начало

    return bio

download_bytes(url, *, chunk_size=DOWNLOAD_CHUNK_SIZE) async

Скачивает файл по URL и возвращает bytes в памяти.

Внимание: весь файл загружается в оперативную память. Не используйте для файлов >100–200 МБ без контроля.

Parameters:

Name	Type	Description	Default
url	str	URL файла.	required
chunk_size	int	Размер чанка при потоковом чтении.	DOWNLOAD_CHUNK_SIZE

Returns:

Name	Type	Description
bytes	bytes	Содержимое файла

Raises:

Type	Description
DownloadFileError	при ошибке скачивания.

Source code in maxapi/connection/base.py

async def download_bytes(
    self,
    url: str,
    *,
    chunk_size: int = DOWNLOAD_CHUNK_SIZE,
) -> bytes:
    """
    Скачивает файл по URL и возвращает bytes в памяти.

    Внимание: весь файл загружается в оперативную память.
    Не используйте для файлов >100–200 МБ без контроля.

    Args:
        url: URL файла.
        chunk_size: Размер чанка при потоковом чтении.

    Returns:
        bytes: Содержимое файла

    Raises:
        DownloadFileError: при ошибке скачивания.
    """
    bio = await self.download_bytes_io(url=url, chunk_size=chunk_size)

    return bio.read()