Ошибка с execute_batch в gql на Python: причина и решение

Разбираем ошибку AttributeError: нет execute_batch в gql при работе с GraphQL на Python через httpx. В какой версии появится и как запустить уже сейчас.

Когда вы подключаете клиент GraphQL на Python поверх httpx, возникает соблазн объединять операции в батчи ради большей пропускной способности и меньшего числа сетевых походов. Частая ловушка проявляется сразу, как только вы пытаетесь вызвать execute_batch у сессии: AttributeError: 'AsyncClientSession' object has no attribute 'execute_batch'. Ниже — короткое объяснение, почему это происходит, и прямой способ все запустить.

Как воспроизвести проблему

Обычно схема такая: создаете Client с HTTPXAsyncTransport, открываете асинхронный контекст и вызываете execute_batch с перечнем объектов GraphQLRequest.

from gql import Client, GraphQLRequest
from gql.transport.httpx import HTTPXAsyncTransport
from itertools import batched
import asyncio

async def run():
    async with Client(
        transport=HTTPXAsyncTransport(url=GQL_URL, headers=GQL_HEADERS, timeout=3000),
        batch_interval=5
    ) as ctx:
        result = await ctx.execute_batch([
            GraphQLRequest(document=QUERY, variable_values=VARIABLES) for chunk in batched(DATA, 1000)
        ])
        print(result)

asyncio.run(run())

Что происходит

API execute_batch недоступен в текущей установленной стабильной ветке библиотеки gql. Судя по релизам проекта, execute_batch появится в версии 4 gql, которая пока в бете. Поэтому объект сессии не имеет этого атрибута и во время выполнения выбрасывает AttributeError.

Как исправить

Если вам окей работать с бета-версией, установите бета-выпуск v4, чтобы получить execute_batch:

pip install gql==v4.0.0b0

После этого тот же клиентский код сможет вызывать execute_batch как задумано.

from gql import Client, GraphQLRequest
from gql.transport.httpx import HTTPXAsyncTransport
from itertools import batched
import asyncio

async def run():
    async with Client(
        transport=HTTPXAsyncTransport(url=GQL_URL, headers=GQL_HEADERS, timeout=3000),
        batch_interval=5
    ) as ctx:
        result = await ctx.execute_batch([
            GraphQLRequest(document=QUERY, variable_values=VARIABLES) for chunk in batched(DATA, 1000)
        ])
        print(result)

asyncio.run(run())

Почему это важно

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

Выводы

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