При web3-разработке часто приходится заниматься вопросами консистентности считываемых из блокчейна данных и оптимизацией таких запросов так как у RPC‑провайдеров достаточно жёсткая тарификация за каждую единицу газа/RPC‑запрос. Обычно достаточно использования Multicall (умный контракт, который объединяет вызовы view‑ и pure‑ функций других умных контрактов в одном RPC‑методе eth_call), однако такое решение имеет свои лимиты. Так Multicall должен умещаться в предельный gaslimit, допускаемый RPC‑провайдером (зачастую он не очень велик и самое большое, что есть на рынке — это всего 500 миллионов единиц газа у одного из публичных RPC‑провайдеров). Другой проблемой чистого multicall является проблема консистентности: дробление запроса по нескольким вызовам multicall, чтобы уложиться в лимиты, может привести к тому, что часть запросов получат данные из текущего блока, а часть из следующего, что по очевидным причинам не подходит, например, для получения состояния всех пулов какого‑нибудь DEX типа Uniswap или Algebra при трейдинге. Крайне элегантное и эффективное решение данной проблемы для python‑разработчиков было выполнено командой YearnFinance (а именно BobTheBuilder) и опубликовано в виде pip‑пакета dank_mids. Именно его мы разберём в этой статье.

Данное решение работает по принципу middleware для Web3-клиента, автоматически собирая вызовы всех view‑ и pure‑ функций умных контрактов в проекте, объединяя их в более крупные вызовы к контракту MulticallV3, которые в свою очередь при достижении указанного лимита по газу на один вызов объединяются в единый batch JSON‑RPC запрос и по HTTP уходят единым запросом к ноде или в эндпоинт RPC‑провайдера, после чего при получении ответа результаты распаковываются в обратном порядке.

Примеры использования dank_mids

Чтобы использовать danks_mids в python‑модуле вам понадобится импортировать его и asyncio инструкцией вида import dank_mids, asyncio.

Далее для каждой вызываемой функции или сгруппированных по смыслу/параметрам функций необходимо написать асинхронную обёртку, принимающую в качестве параметра экземпляр dank_mids.Contract, который во многом эквивалентен классам Contract из web3py или ProjectContract из brownie и может быть получен из них либо прямым привидением типа, либо как результат вызова функции dank_mids.patch_contract(contract: Contract). Например:

# Пример патчинга контракта из Brownie
contract = Contract.from_abi(name, address, abi, persist=False)
patched_contract = dank_mids.patch_contract(contract)

# Либо просто по адресу, если контракт верифицирован на etherscan или его экземпляре для отдельной сети
dank_mids_contract = dank_mids.Contract(contract_address)

И примеры асинхронных функций обёрток:

# Вызов нескольких функций без параметров из одного контракта.
# В данном случае получение адресов токенов для пулов Uniswap/Algebra.
async def get_tokens_for_pool(pool: dank_mids.Contract):
  return await asyncio.gather(pool.token0, pool.token1)

# Вызов функции с параметром, например, получение баланса аккаунта в токене.
async def get_balance_of(token: dank_mids.Contract, address: str):
    return await asyncio.gather(token.balanceOf.coroutine(address))

# Вызов батча одинаковых функций с параметрами из массива.
async def get_balances(token: dank_mids.Contract, addresses: list[str]):
    return await asyncio.gather(*(token.balanceOf.coroutine(address) for address in addresses)) 

# Вызов функции, которая может упасть с исключением.
# В этом случае среди результатов нужно будет отлавливать объекты класса Exception.
async def div(math: dank_mids.Contract, a : int, b : int):
  return await asyncio.gather(math.div.coroutine(a, b), return_exceptions=True)

# Вызов одной и той же функции в разных блоках.
# Может быть очень полезно при сборе статистики или при обработке реорганизаций блоков.
async def get_balances_in_different_blocks(token: dank_mids.Contracts, address: str, blocks: list[int]):
    return await asyncio.gather(*(token.balanceOf.coroutine(address, block_identifier=block) for block in blocks))

После написания всех асинхронных функций‑обёрток необходимо собрать их вызовы вместе, вызвать и распарсить результаты. В этом случае будет использован ровно один rpc‑запрос к ноде/провайдеру как в немного упрощённом примере из самого проекта dank_mids:

blocks = [15_000_000, 15_100_000, 15_200_000, 15_300_000, 15_400_000, 15_500_000]
uniswap_pools = [
    "0xBb2b8038a1640196FbE3e38816F3e67Cba72D940",
    "0x0d4a11d5EEaaC28EC3F61d100daF4d40471f1852",
    "0xB4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc",
    "0xA478c2975Ab1Ea89e8196811F51A7B7Ade33eB11",
]

def main() -> None:
  asyncio.run(_main())

async def _main() -> None:
  # Получаем экземпляры dank_mids.Contract из адресов контрактов.
  dank_pool_contracts = tuple(map(dank_mids.Contract, uniswap_pools))
  # Получаем экземпляр dank_mids.Contracts для токена WETH
  token = dank_mids.Contract("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2")
  address = "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045" # vitalik.eth (адрес Бутерина)

  tokens, balances = await asyncio.gather(
        asyncio.gather(*(get_tokens_for_pool(pool) for pool in dank_pool_contracts)),
        asyncio.gather(get_balances_in_different_blocks(token, address, blocks)),
        return_exceptions=True # Если мы ожидаем, что какой-то из вызовов может упасть с исключением
    )

  print(tokens)
  print(balances)

Важно понимать, что это не единственный способ вызова функций через dank_mids: можно также вызывать всё в разных местах по отдельности, но тогда нужно будет настроить продолжительность периодов вызова multicall через env (в течение указанного периода dank_mids middleware собирает запросы на вызов функций со всего проекта, после чего вызывает их одновременно и возвращает результаты в те участки кода, что их ожидают),‑ такой подход может быть более удобен в legacy‑проектах.

Перевод проекта на использование dank_mids и eth‑brownie

В общем случае у вас есть два существующих репозитория, один из которых содержит используемые вами умные контракты (если у вас есть собственные, далее cotracts‑repo), а другой содержит код с бекендом/микросервисом на python, использующий стандартный web3py (далее backend‑repo). Чтобы быстро начать использовать dank_mids через brownie необходимо сделать следующие шаги:

Установить brownie и dank_mids (либо добавить их в requirements.txt/pyproject.toml и далее через poetry или uv):
pip install eth-brownie dank-mids
Инициализировать новый brownie‑проект в новой директории:
brownie init new-directory-for-your-project
Подтянуть contracts‑repo подмодулем в директорию contracts в brownie‑проекте, после чего собрать его через brownie compile (может не заработать с первой попытки если у вас много зависимостей, в этом случае вам нужно будет настроить remappings в brownie‑config.yaml и задать такие же настройки компилятора, как в вашем исходном проекте). С проектами на foundry, wake, ape, hardhat и mocassin обычно проблем не бывает и всё работает из коробки или с точно теми же remappings и настройками компилятора, а если что‑то не работает, то проблемы как правило минорные и с ними даже LLM‑агент справляется.
Подтянуть backend‑repo сабмодулем в директорию scripts в brownie‑проекте, после чего можно будет запустить ваш сервис командой brownie run main/app/mamange --network network_name.
Немного поправить кодовую базу, чтобы view и pure функции в проекте вызывались через dank_mids. В этом вам может быть крайне полезен пример использования от автора из репозитория проекта.
Готово, можете запускать теперь ваш проект через brownie run и наслаждаться быстродействием и сниженным потреблением ресурсов.

Вообще запуск через brownie — это не единственный способ использования dank_mids, но это самый просто способ попробовать на мой взгляд. Впрочем так как dank_mids это полностью совместимый с web3py middleware, то можно заставить ваш проект работать с ним напрямую через добавление middleware по документации web3py, но может потребоваться некоторое время на это. Если будут возникать сложности с этим, то пишите в комментарии и мы разберём вместе с коллегами по цеху возникшие проблемы и вопросы.

Ограничения

У данного решения есть те же самые ограничения, что и у обычного MulticallV3. Например, невозможность задать кастомный msg.sender для view‑функции, которая имеется при вызове с помощью прямого eth_call как в следующем примере из документации web3py:

token_contract.functions.myBalance().call({'from': web3.eth.accounts[0]})
12345  # the token balance for `web3.eth.accounts[0]`

В случае использования MulticallV3 или dank_mids поле msg.sender всегда будет адресом контракта multicall и явно задать возможно только tx.origin, который по умолчанию в dank_mids да и в web3py устанавливается равным address(0). Иногда встречаются view‑функции, использующие в работе msg.sender и иногда даже ограничивающие доступ при вызове по нему и адекватного решения типа MulticallV3 или dank_mids до сих пор нет и приходится вручную оборачивать батч eth_call с такими функциями в JSON‑RPC, а потом расспрашивать ответы, отделяя их от исключений.

Другим ограничением является невозможность использования dank_mids на стандартной тестовой сети development в brownie, wake или же ape,‑ локальная тестовая сеть как правило сильно обвешана различными middleware для web3py, среди которых всегда найдётся какой‑нибудь, с которым конфликтует dank_mids, который тоже сам по себе выполнен в формате middleware для web3py. Эта проблема может возникать и для некоторых POA‑сетей, использующих какой‑нибудь очень специфический POA_middleware, поэтому при работе с ними приходится отключать некоторые дефолтные middleware.

Третьим ограничением является используемая версия python: не смотря на то, что dank_mids свежий и регулярно обновляемый пакет, он не поддерживает python 3.14 и выше, а также режим с выключенным GIL (даже не смотря на то, что это уже де‑факто стандарт на самых последних ubuntu‑серверах версии 26.04 LTS, поэтому приходится крутить прод либо на более старых версиях ubuntu‑сервера, либо на других дистрибутивах GNU/Linux, либо даже на windows server).

Но banteg из YearnFinance всё равно одобряет

Заключение

В заключение хочется сказать только одно: пишите хороший и эффективный код, а плохой не пишите. Это не экономия на спичках: использование dank_mids может серьёзно уменьшить latency вашего прода, сократить объём кодовой базы, сделав её чище и понятнее, а также существенно сократить издержки на инфраструктуру, сэкономив вам тысячи, а то и десятки тысяч долларов в месяц только на серверных мощностях (всё таки почти чистый Си вместо интерпретируемого python под капотом) и потребляемом JSON‑RPC over HTTP (за счёт использования multicall в связке с rpc batching).

В одной из следующих статей разберу как без особых трудозатрат перевести web3 python проект на faster‑web3-py стек, снизив latecy, увеличив скорость работы и снизив потребление CPU за счёт того, что большая часть python‑кода будет работать через компиляцию в Си через mypyc вместо классического подхода через python‑интерпретатор. А также как заставить работать Multicall в режиме аналогичном dank_mids в библиотеке Nethereum, регулярным контрибьютором которой я являюсь, так что подписывайтесь на хабре, если интересна тема оптимизации кода при работе с web3.

Как жить когда не хватает Multicall