BaseConnection

BaseConnection()

Bases: BotMixin

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

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

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

Атрибуты

bot (Optional[Bot]): Экземпляр бота. session (Optional[ClientSession]): aiohttp-сессия. after_input_media_delay (float): Задержка после ввода медиа.

Source code in maxapi/connection/base.py

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

    Атрибуты:
        bot (Optional[Bot]): Экземпляр бота.
        session (Optional[ClientSession]): aiohttp-сессия.
        after_input_media_delay (float): Задержка после ввода медиа.
    """

    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 (str): Новый 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 (HTTPMethod): HTTP-метод (GET, POST и т.д.).
        path (ApiPath | str): Путь до конечной точки.
        model (BaseModel | Any, optional): Pydantic-модель для
            десериализации ответа, если is_return_raw=False.
        is_return_raw (bool, optional): Если True — вернуть сырой
            ответ, иначе — результат десериализации.
        **kwargs: Дополнительные параметры (query, headers, json).

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

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

    bot = self._ensure_bot()

    if not bot.session:
        bot.session = ClientSession(
            base_url=bot.api_url,
            timeout=bot.default_connection.timeout,
            headers=bot.headers,
            **bot.default_connection.kwargs,
        )

    conn = bot.default_connection
    max_retries = conn.max_retries
    retry_statuses = conn.retry_on_statuses
    backoff_factor = conn.retry_backoff_factor

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

    for attempt in range(max_retries + 1):
        try:
            r = await bot.session.request(
                method=method.value,
                url=url,
                **kwargs,
            )
        except ClientConnectionError as e:
            if attempt < max_retries:
                delay = backoff_factor * (2**attempt)
                logger_bot.warning(
                    f"Ошибка соединения ({e}), "
                    f"попытка {attempt + 1}/{max_retries + 1}, "
                    f"жду {delay:.1f}с"
                )
                await asyncio.sleep(delay)
                continue
            raise MaxConnection(f"Ошибка при отправке запроса: {e}") from e

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

        if r.status in retry_statuses and attempt < max_retries:
            await r.read()
            delay = backoff_factor * (2**attempt)
            logger_bot.warning(
                f"Серверная ошибка {r.status}, "
                f"попытка {attempt + 1}/{max_retries + 1}, "
                f"жду {delay:.1f}с"
            )
            await asyncio.sleep(delay)
            continue

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

        break

    raw = await r.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

    if hasattr(model, "message"):
        attr = model.message
        if hasattr(attr, "bot"):
            attr.bot = bot

    if hasattr(model, "bot"):
        model.bot = bot  # type: ignore

    return model

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 (str): URL загрузки.
        path (str): Путь к файлу.
        type (UploadType): Тип файла.

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

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

    path_object = Path(path)
    basename = path_object.name
    ext = path_object.suffix

    form = FormData(quote_fields=False)
    form.add_field(
        name="data",
        value=file_data,
        filename=basename,
        content_type=f"{type.value}/{ext.lstrip('.')}",
    )

    bot = self._ensure_bot()

    session = bot.session
    if session is not None and not session.closed:
        response = await session.post(url=url, data=form)
        return await response.text()
    else:
        async with ClientSession() as temp_session:
            response = await temp_session.post(url=url, data=form)
            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 (str): Имя файла.
        url (str): URL загрузки.
        buffer (bytes): Буфер данных.
        type (UploadType): Тип файла.

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

    try:
        matches = puremagic.magic_string(buffer[:4096])
        if matches:
            mime_type = matches[0][1]
            ext = mimetypes.guess_extension(mime_type) or ""
        else:
            mime_type = f"{type.value}/*"
            ext = ""
    except Exception:
        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:
        response = await session.post(url=url, data=form)
        return await response.text()
    else:
        async with ClientSession() as temp_session:
            response = await temp_session.post(url=url, data=form)
            return await response.text()