Web3.0 на Python, часть 2: advanced / forpes.ru

Главная
Web3.0 на Python, часть 2: advanced

Web3.0 на Python, часть 2: advanced +4

24.11.2022 13:34

shamaevnn 0 1600 Источник

Привет, хабр! В первой части мы рассмотрели базовые операции на web3py, которые закроют большинство ваших потребностей для проектов на ранних этапах. Здесь же речь в основном пойдет про улучшение производительности и различные "фишки", которые, например, помогут вам уменьшить количество запросов и/или время на эти запросы. Скорее всего, они не будут полезны тем, кто делает какой-то pet-project или проект на хакатоне. А полезны они будут тем, кто делает реальный боевой проект и кому важна производительность.

Содержание:

Оценка газа ⛽

Мы уже делали транзакцию с нативной валютой, ставя при этом "газ побольше", чтоб точно хватило. А что если хочется показать пользователю заранее, сколько именно он должен будет потратить газа на транзакцию? В данном случае нам поможет метод eth.estimate_gas. Стоит сказать, что это число подвержено флуктуации. И если вдруг предсказанного газа не хватит, то транзакция зафейлится, а пользователь потеряет какое-то количество нативной валюты. Поэтому я бы советовал увеличивать это количество на ~20%, неиспользованный газ просто не будет потрачен.

from web3 import Web3

user_address = "0x2A647559a6c5dcB76ce1751101449ebbC039b157"
rpc_url = "https://matic-mumbai.chainstacklabs.com"  # testnet Polygon
web3 = Web3(Web3.HTTPProvider(rpc_url))

amount_of_matic_to_send = 1
txn = {  # формируем транзакцию
  'chainId': web3.eth.chain_id,
  'from': user_address,
  'to': user_address,
  'value': int(Web3.toWei(amount_of_matic_to_send, 'ether')),
  'nonce': web3.eth.getTransactionCount(user_address), 
  'gasPrice': web3.eth.gas_price,
}
web3.eth.estimate_gas(txn)
# предсказанное количество газа 21000 Wei

Ускоряем инициализацию контрактов

Представим, что нам по адресу ERC-20 токена и адресу юзера нужно узнать его (юзера) баланс. Напомню, как мы делали это в первой части:

import json
from web3 import Web3

# одинаковый для всех ERC20 токенов
ERC20_ABI = json.loads('''[{"inputs":[{"internalType":"string","name":"_name","type":"string"},{"internalType":"string","name":"_symbol","type":"string"},{"internalType":"uint256","name":"_initialSupply","type":"uint256"}],"stateMutability":"nonpayable","type":"constructor"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"owner","type":"address"},{"indexed":true,"internalType":"address","name":"spender","type":"address"},{"indexed":false,"internalType":"uint256","name":"value","type":"uint256"}],"name":"Approval","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"from","type":"address"},{"indexed":true,"internalType":"address","name":"to","type":"address"},{"indexed":false,"internalType":"uint256","name":"value","type":"uint256"}],"name":"Transfer","type":"event"},{"inputs":[{"internalType":"address","name":"owner","type":"address"},{"internalType":"address","name":"spender","type":"address"}],"name":"allowance","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"stateMutability":"view","type":"function"},{"inputs":[{"internalType":"address","name":"spender","type":"address"},{"internalType":"uint256","name":"amount","type":"uint256"}],"name":"approve","outputs":[{"internalType":"bool","name":"","type":"bool"}],"stateMutability":"nonpayable","type":"function"},{"inputs":[{"internalType":"address","name":"account","type":"address"}],"name":"balanceOf","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"decimals","outputs":[{"internalType":"uint8","name":"","type":"uint8"}],"stateMutability":"view","type":"function"},{"inputs":[{"internalType":"address","name":"spender","type":"address"},{"internalType":"uint256","name":"subtractedValue","type":"uint256"}],"name":"decreaseAllowance","outputs":[{"internalType":"bool","name":"","type":"bool"}],"stateMutability":"nonpayable","type":"function"},{"inputs":[{"internalType":"address","name":"spender","type":"address"},{"internalType":"uint256","name":"addedValue","type":"uint256"}],"name":"increaseAllowance","outputs":[{"internalType":"bool","name":"","type":"bool"}],"stateMutability":"nonpayable","type":"function"},{"inputs":[],"name":"name","outputs":[{"internalType":"string","name":"","type":"string"}],"stateMutability":"view","type":"function"},{"inputs":[{"internalType":"uint8","name":"decimals_","type":"uint8"}],"name":"setupDecimals","outputs":[],"stateMutability":"nonpayable","type":"function"},{"inputs":[],"name":"symbol","outputs":[{"internalType":"string","name":"","type":"string"}],"stateMutability":"view","type":"function"},{"inputs":[],"name":"totalSupply","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"stateMutability":"view","type":"function"},{"inputs":[{"internalType":"address","name":"recipient","type":"address"},{"internalType":"uint256","name":"amount","type":"uint256"}],"name":"transfer","outputs":[{"internalType":"bool","name":"","type":"bool"}],"stateMutability":"nonpayable","type":"function"},{"inputs":[{"internalType":"address","name":"sender","type":"address"},{"internalType":"address","name":"recipient","type":"address"},{"internalType":"uint256","name":"amount","type":"uint256"}],"name":"transferFrom","outputs":[{"internalType":"bool","name":"","type":"bool"}],"stateMutability":"nonpayable","type":"function"}]''')

# USDT токен
usdt_contract_address = '0xA11c8D9DC9b66E209Ef60F0C8D969D3CD988782c'

def get_balance_old(web3: Web3, token_address: str, user_address: str) -> int:
    # инициализация контракта
    token_contract = web3.eth.contract(token_address, abi=ERC20_ABI)

    # получение данных из ноды
    balance = token_contract.functions.balanceOf(user_address).call()
    return int(balance)

Напрашивается очевидный вопрос: нам нужна всего лишь одна функция balanceOf, но почему мы при этом инициализируем контракт со всеми функциями из ERC20_ABI? Да, можно просто вручную убрать все остальные функции из ABI, чтобы в итоге остался список из одной нужной функции. А можно сделать и вот так (такой способ пригодится в дальнейшем):

import json
from web3.types import ABIFunction
from eth_utils import encode_hex, function_abi_to_4byte_selector, add_0x_prefix
from web3._utils.contracts import encode_abi
from web3._utils.abi import get_abi_output_types


encode_hex_fn_abi = lambda fn_abi: encode_hex(
    function_abi_to_4byte_selector(fn_abi)
)
# словарь с функцией balanceOf. Можно посмотреть тут https://bscscan.com/token/0x8ac76a51cc950d9822d68b83fe1ad97b32cd580d#code
BALANCE_OF_ABI: ABIFunction = json.loads('{"constant":true,"inputs":[{"internalType":"address","name":"account","type":"address"}],"name":"balanceOf","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"}'
balance_of_output_types = get_abi_output_types(BALANCE_OF_ABI)
balance_of_selector = encode_hex_fn_abi(BALANCE_OF_ABI)

def get_balance_new(web3: Web3, user_address: str, token_address: str) -> int:
    # инициализация контракта
    data = add_0x_prefix(
        encode_abi(
            web3, 
            abi=BALANCE_OF_ABI,
            arguments=(user_address,), # аргументы функции balanceOf
            data=balance_of_selector
        ),
    )
    
    tx = {"to": token_address, "data": data}
    # получение данных из ноды
    res = web3.eth.call(tx)
    
    output_data = web3.codec.decode_abi(balance_of_output_types, res)
    
    balance = output_data[0]
    return balance

Сравним, что же у нас получается по времени CPU и Wall Time (WT) в ms, усреднённо:

	old WT	old CPU	new WT	new CPU
Инициализация	16	18	3	3
Вызов ноды	155	19.5	145	17

Разница не велика, скажете вы, порядка 10 ms. А я скажу вам, что тут десяток, там два десятка и вот он, прирост. И вообще, с миру по нитке ????

Баланс нескольких токенов за один запрос ⚡

Допустим, мы хотим посмотреть баланс USDT, USDC и других токенов у юзера. Для этого придется делать N запросов с функцией balanceOf, где N — количество токенов. К счастью, для некоторых сетей, а именно:

Ethereum
Ethereum Rinkeby
Ethereum Kovan
Binance Smart Chain
Polygon

есть так называемые контракты BalanceScanner, адреса можно найти тут, а интерфейс взаимодействия можно глянуть тут. В них есть две очень полезные нам функции:

tokensBalance — узнать баланс нескольких токенов для юзера;
tokenBalances — узнать баланс нескольких юзеров для токена.

В качестве примера рассмотрим использование первой функции tokensBalance.

from eth_utils import encode_hex, function_abi_to_4byte_selector, add_0x_prefix
from web3._utils.contracts import encode_abi
from web3._utils.abi import get_abi_output_types
from web3 import Web3
from web3.types import HexBytes

# инициализация всего и вся
web3 = Web3(Web3.HTTPProvider("https://bsc-dataseed1.defibit.io"))

# ABI функции tokensBalance. Взять можно тут https://bscscan.com/address/0x83cb147c13cBA4Ba4a5228BfDE42c88c8F6881F6#code
TOKENS_BALANCE_ABI = {"inputs":[{"internalType":"address","name":"owner","type":"address"},{"internalType":"address[]","name":"contracts","type":"address[]"}],"name":"tokensBalance","outputs":[{"components":[{"internalType":"bool","name":"success","type":"bool"},{"internalType":"bytes","name":"data","type":"bytes"}],"internalType":"struct BalanceScanner.Result[]","name":"results","type":"tuple[]"}],"stateMutability":"view","type":"function"}
TOKENS_BALANCE_SELECTOR = encode_hex(function_abi_to_4byte_selector(TOKENS_BALANCE_ABI))
tokens_balance_output_types = get_abi_output_types(TOKENS_BALANCE_ABI)

# баланс каких токенов будем проверять
tokens_to_check = [
    '0x55d398326f99059fF775485246999027B3197955', # USDT
    '0x8AC76a51cc950d9822D68b83fE1Ad97B32Cd580d', # USDC
    '0xe9e7CEA3DedcA5984780Bafc599bD69ADd087D56', # BUSD
]
# адрес пользователя, для которого будем смотреть баланс
user_address = "0x2A647559a6c5dcB76ce1751101449ebbC039b157"
encoded_data = encode_abi(
    web3=web3,
    abi=TOKENS_BALANCE_ABI,
    arguments=(user_address, [t for t in tokens_to_check]),  # аргументы функции tokensBalance
    data=TOKENS_BALANCE_SELECTOR,
)
tx = {
  "to": "0x83cb147c13cBA4Ba4a5228BfDE42c88c8F6881F6",  # адрес контракта BalanceScanner
  "data": encoded_data
}
# обращаемся к ноде
tx_raw_data = web3.eth.call(tx)
output_data = web3.codec.decode_abi(tokens_balance_output_types, tx_raw_data)[0]
res = {}
for token_address, (_, bytes_balance) in zip(tokens_to_check, output_data):
    wei_balance = web3.codec.decode_abi(["uint256"], HexBytes(bytes_balance))[0]
    res[token_address] = wei_balance
# в res получаем словарь, где ключ - адрес токена, значение - баланс
#{
#    '0x55d398326f99059fF775485246999027B3197955': 19357349465782901200,
#    '0x8AC76a51cc950d9822D68b83fE1Ad97B32Cd580d': 38062949201715000000,
#    '0xe9e7CEA3DedcA5984780Bafc599bD69ADd087D56': 0
#}

Еще раз: баланс сразу нескольких токенов/юзеров мы можем узнать всего лишь за один запрос к ноде, круто же? Просмотр баланса определенного токена для нескольких юзеров c помощью функции tokenBalances оставим в качестве домашнего задания.

Его величество multicall ⚡⚡

Контракты BalanceScanner объединяют в себе несколько вызовов других контрактов, но они заточены только для просмотра баланса. Multicall контракты же позволяют объединить вызовы произвольных контрактов. Список multicall контрактов для сетей можно найти тут.

В качестве примера: хотим понять, сколько allowance дал пользователь различным контрактам на разные токены (что такое allowance и зачем он нужен).

from dataclasses import dataclass
from typing import Dict, List
from eth_utils import encode_hex, function_abi_to_4byte_selector, add_0x_prefix
from web3._utils.contracts import encode_abi
from web3._utils.abi import get_abi_output_types
from web3 import Web3


@dataclass
class ApproveAddressesInfo:
    approval_address: str  # контракт, которому дали approve
    token_address: str  # адрес токена

encode_hex_fn_abi = lambda fn_abi: encode_hex(
    function_abi_to_4byte_selector(fn_abi)
)

# можно подсмотреть в ABI ERC-20 
ALLOWANCE_ABI = {"constant": True, "inputs": [{"internalType": "address", "name": "owner", "type": "address"}, {"internalType": "address", "name": "spender", "type": "address"}], "name": "allowance", "outputs": [{"internalType": "uint256", "name": "", "type": "uint256"}], "payable": False, "stateMutability": "view", "type": "function"}
allowance_selector = encode_hex_fn_abi(ALLOWANCE_ABI)
# а это в ABI multicall контрактов
TRY_AGGREGATE_ABI = {"inputs":[{"name":"requireSuccess","type":"bool"},{"components":[{"name":"target","type":"address"},{"name":"callData","type":"bytes"}],"name":"calls","type":"tuple[]"}],"name":"tryAggregate","outputs":[{"components":[{"name":"success","type":"bool"},{"name":"returnData","type":"bytes"}],"name":"returnData","type":"tuple[]"}],"stateMutability":"nonpayable","type":"function"}
try_aggregate_selector = encode_hex_fn_abi(TRY_AGGREGATE_ABI)
try_aggregate_output_types = get_abi_output_types(TRY_AGGREGATE_ABI)


def multicall_check_allowance(
    web3: Web3,
    multicall_address: str,
    user_address: str,
    approval_token_addresses: List[ApproveAddressesInfo],
) -> Dict[ApproveAddressesInfo, str]:
    encoded = (
        (
            addresses.token_address,
            encode_abi(
                web3,
                ALLOWANCE_ABI,
                # адрес пользователя и адрес контракта на approve
                arguments=(sender_address, addresses.approval_address),
                data=allowance_selector,
            ),
        )
        for addresses in approval_token_addresses
    )
    
    data = add_0x_prefix(
        encode_abi(
            web3,
            TRY_AGGREGATE_ABI,
            (False, [(token_addr, enc) for token_addr, enc in encoded]),
            try_aggregate_selector,
        )
    )
    tx_raw_data = web3.eth.call({"to": multicall_address, "data": data})
    output_data = web3.codec.decode_abi(try_aggregate_output_types, tx_raw_data)[0]
    output_data = (
        str(decode_abi(["uint256"], HexBytes(raw_token_address))[0]) for (_, raw_token_address) in output_data
    )
    return dict(zip(approval_token_addresses, output_data))

Пример вызова этой функции:

multicall_check_allowance(
    ...,
    user_address='0x...',
    # список из адреса токенов и адреса контракта,
    # для которого нужно посомтреть allowance
    approval_token_addresses=[
        ApproveAddressesInfo(
            approval_address="0x24ED43C718714eb63d5aA57B78B54704E256024E",
            token_address="0x55d398326f99059fF775485246999027B3197955",
        ),
        ApproveAddressesInfo(
            approval_address="0x51FC43C718714eb63d5aA57B78B54704E256024E",
            token_address="0x8AC76a51cc950d9822D68b83fE1Ad97B32Cd580d",
        )                    
    ]
)
# На выходе для каждой пары получим число — сколько юзер разрешил контракту снимать токенов
"""
ApproveAddressesInfo(
    approval_address="0x24ED43C718714eb63d5aA57B78B54704E256024E",
    token_address="0x55d398326f99059fF775485246999027B3197955",
): 131231231231,
ApproveAddressesInfo(
    approval_address="0x51FC43C718714eb63d5aA57B78B54704E256024E",
    token_address="0x8AC76a51cc950d9822D68b83fE1Ad97B32Cd580d",
): 0
"""

И еще раз: за один вызов RPC ноды мы получили allowance пользователя сразу для нескольких различных пар токенов и контрактов. Думаю, что если вы делаете какой-то web3 сервис, то точно найдете применение для multicall. А учитывая тот факт, что многие сервисы по предоставлению RPC нод платные, мы сильно экономим $ на запросах.

Асинхронный web3 ????

На дворе уже как минимум 2022 год. Это значит, что пора использовать все прелести асинхронного кода и в web3py. Сравним скорость на простом примере: получение gas_price для нескольких сетей (в данном случае 6):

import time
import asyncio
from web3 import Web3
from web3.eth import AsyncEth
from typing import Dict, List

rpc_url_by_chain_id = {
    56: "https://bsc-dataseed3.ninicoin.io", 137: "https://rpc-mainnet.matic.quiknode.pro",
    250: "https://rpc.fantom.network", 43114: "https://rpc.ankr.com/avalanche",
    42161: "https://rpc.ankr.com/arbitrum", 10: "https://mainnet.optimism.io", 
}

# инициализация СИНХРОННОЙ версии web3 для каждой сети 
sync_web3_by_chaid_id = {}
for chain_id, rpc_url in rpc_url_by_chain_id.items():
    sync_web3_by_chaid_id[chain_id] = Web3(Web3.HTTPProvider(rpc_url))

# инициализация АСИНХРОННОЙ версии web3 для каждой сети 
async_web3_by_chaid_id = {}
for chain_id, rpc_url in rpc_url_by_chain_id.items():
    async_web3 = Web3(
        Web3.AsyncHTTPProvider(rpc_url), 
        modules={"eth": (AsyncEth,)}, middlewares=[]
    )
    async_web3_by_chaid_id[chain_id] = async_web3

def get_sync(chain_ids: List[int]) -> Dict[int, int]:
    """
    chain_ids = [56,137,250,43114,42161,10] -> {1: 100_000, 56: 200_000, ...}
    """
    res, start_time = {}, time.time()
    for chain_id in chain_ids:
        web3 = sync_web3_by_chaid_id[chain_id]
        res[chain_id] = web3.eth.gas_price
    print(f'{(time.time()-start_time):.3f} seconds for SYNC version')
    return res    
  
async def get_async(chain_ids: List[int]) -> Dict[int, int]:
    """
    chain_ids = [56,137,250,43114,42161,10] -> {1: 100_000, 56: 200_000, ...}
    """
    tasks, res, start_time = [], {}, time.time()
    for chain_id in chain_ids:
        async_web3 = async_web3_by_chaid_id[chain_id]
        tasks.append(async_web3.eth.gas_price)  # добавляем асинхронные задачи
    gas_prices = await asyncio.gather(*tasks)  # запускаем сразу несколько задач      
    print(f'{(time.time()-start_time):.3f} seconds for ASYNC version')
    return dict(zip(chain_ids, gas_prices))
            
    
if __name__ == "__main__":
    get_sync([56,137,250,43114,42161,10])
    asyncio.run(get_async([56,137,250,43114,42161,10]))
    
# (в среднем на 5 запусках)
# 2.391 seconds for SYNC version
# 0.493 seconds for ASYNC version

Разница почти в 5 раз, неплохо, да? В целом, если вы начинаете проект в 2022+ году с web3py, я бы посоветовал вам использовать именно асинхронную версию. Но у этого есть и подводные камни. Например, если мы хотим узнать баланс нативной валюты для N пользователей, можно написать что-то в духе:

tasks = []
for user_address in user_addresses:
    tasks.append(web3.eth.get_balance(user_address))
res = asyncio.gather(*tasks)

Конечно, в этом случае это также будет намного быстрее синхронного варианта. Но! В этом случае мы почти одновременно делаем N запросов к ноде. Во-первых, наши запросы = деньги, поэтому стоит их экономить, т.е. делать меньше запросов. Во-вторых, не удивляйтесь, что из-за такого "дудоса" нода будет отвечать дольше или не отвечать вовсе. Нужно быть внимательным в этом плане. Например, использовать BalanceScanner или Multicall, чтобы забирать все одним запросом.

Полезные сервисы ????

Очевидно, что, делая web3-продукт, невозможно обойтись использованием только web3py. На помощь приходят и другие сервисы.

Провайдеры блокчейн нод. Конечно, на самых-самых ранних этапах проекта можно обойтись и бесплатными RPC, например, с chainlist.org. Но в какой-то момент вы заметите, что этого будет не хватать: ответ будет идти дольше чем нужно или вообще не будет приходить. Есть вариант поднимать свою ноду для каждого блокчейна, но, почти наверное, вы потратите намного больше человеко-часов на ее поднятие и поддержку. Поэтому рекомендуется использовать платные сервисы, которые все делают за вас.
Цены токенов. У юзера есть портфель из токенов, хочется посчитать общую сумму. Вам нужен сервис, позволяющий узнать цену каждого токена. По факту, многие используют CoinGecko. Также, из этого списка можно выделить chain.link, т.к. этот сервис позволяет узнавать цену напрямую с помощью вызова функций контракта.
Блокчейн-эксплореры (сканнеры). Позволяют узнать информацию о сущностях блокчейна: детали транзакции, текущий газ, все транзакции юзера и т.д.
Баланс-эксплореры. У юзера N токенов, хочется понять его баланс в $. Конечно, можно сделать N запросов по цене этих токенов к сервисам из пункта 3., перемножить, сложить и получить желаемое, а можно воспользоваться сервисами из данного пункта.
Токен листы. Бывает так, что у юзера много скам-токенов, с которыми практически ничего нигде нельзя сделать. Но как такие токены отфильтровать? Для этого нам и нужны токен листы от проверенных провайдеров. К счастью, есть уже единый агрегированный список от множества провайдеров.

Вот и всё!

Надеюсь, вы узнали что-то новое для себя, а эти примеры будут полезны для ваших проектов: уменьшат количество запросов, сэкономят секунды и $.

Никита Шамаев

Залетайте в телеграм-канал. Там больше постов про разработку/технологии/датасеты, и там вы не пропустите анонс следующей статьи ????