如何在web3.py中進行批量請求？

1. 這是我基于 如何在 web3py 中使用 AsyncHTTPProvider？ 文章。但是，在運行此代碼時，它會像同步函式一樣執行。

2. 對于 web3.js，支持批量請求 https://dapp-world.com/smartbook/web3-batch-request-Eku8。但是，web3.py 沒有。

3. 我正在使用 Ethereum Alchemy API，它每秒支持大約 19 個 API 呼叫。

4. 我有大約 1000 個以太坊地址

5. 如何修改代碼以使每秒能夠批處理 19 個地址？

from web3 import Web3
from web3.eth import AsyncEth
import time 
import pandas as pd
import aiohttp
import asyncio

alchemy_url = "https://eth-mainnet.g.alchemy.com/v2/zCTn-wyjipF5DvGFVNEx_XqCKZakaB57"
w3 = Web3(Web3.AsyncHTTPProvider(alchemy_url), modules={'eth': (AsyncEth,)}, middlewares=[])

start = time.time()

df = pd.read_csv('Ethereum/ethereumaddresses.csv') 
Wallet_Address=(df.loc[:,'Address'])
#Balance_storage = []

session_timeout = aiohttp.ClientTimeout(total=None)


async def get_balances():
    for address in Wallet_Address:
        balance = await w3.eth.get_balance(address)
        print(address, balance)
            
asyncio.run(get_balances())
end = time.time()
total_time = end - start
print(f"It took {total_time} seconds to make {len(Wallet_Address)} API calls")

uj5u.com熱心網友回復：

我認為我的想法不是最好的，但您可以將其用作臨時解決方案。

為此，您必須使用 ThreadpoolExecutor。

我執行了一個基準測驗并發現了這些結果：

• 沒有ThreadpoolExecutor，使用BSC Public RPC，只是在for回圈中運行，需要3分鐘以上才能完成這個程序。 單擊此處查看測驗 1 的輸出

• 使用 ThreadpoolExecutor、BSC 公共 RPC 和 100ms 延遲使用time.sleep(0.1)，在不到 40 秒內完成，如下圖所示。單擊此處查看測驗 2 的輸出

• 使用 ThreadpoolExecutor，使用 Quicknode 和 100ms 延遲，在 35 秒內完成。單擊此處查看測驗 3 的輸出

做簡單的數學運算（1000 個錢包/每秒 19 次呼叫），我們知道您的程序至少需要接近 50 秒。嘗試以 100 毫秒的延遲運行，如果它不起作用，您可以增加更多延遲。

使用的問題之一time.sleep是，如果您使用的是 GUI 或類似的東西，我們無法在此程序中暫停（因為 GUI 將凍結）。（我認為你可以使用多處理來繞過這個 xD）

第二個問題是這樣做可能會改變每個地址在 CSV 中的位置。（您可以為每個地址分配 _id 或類似的屬性，以便在結束后使用 For 回圈進行組織。）

代碼：在 BSC 上作業得很好（只需更改 RPC）。此代碼將找到所有余額并將它們存盤在self.data (defaultdict). 之后，將其保存在名為“newBalances.csv”的新 CSV 檔案中（您可以更改它）

from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from web3 import Web3
import pandas as pd
import time

class multiGetBalanceExample():

    def __init__(self):
        self.initialtime = datetime.now() #initial time

        #=== Setup Web3 ===#
        self.bsc = "https://bsc-dataseed.binance.org/" #rpc (change this)
        self.web3 = Web3(Web3.HTTPProvider(self.bsc)) #web3 connect 

        #=== Loading Csv file ===#
        self.df = pd.read_csv(r"./Ethereum/ethereumaddresses.csv") 
        self.wallet_address=(self.df.loc[:,'Address'])

        #=== Setup Temporary Address/Balance Save Defaultdict ===#
        self.data = defaultdict(list)
        
        #=== Start ===#
        self.start_workers(self.data)

        #=== Finish ===#
        self.saveCsv() #saving in new csv file
        self.finaltime = datetime.now() #end time
        print(f"\nFinished! Process takes: {self.finaltime - self.initialtime}")



    def start_workers(self, data, workers=10):
        with ThreadPoolExecutor(max_workers=workers) as executor:
            [executor.submit(self.getBalances, _data=data,  _from=0, _to=101)]
            [executor.submit(self.getBalances, _data=data,  _from=101, _to=201)]
            [executor.submit(self.getBalances, _data=data,  _from=201, _to=301)]
            [executor.submit(self.getBalances, _data=data,  _from=301, _to=401)]
            [executor.submit(self.getBalances, _data=data,  _from=401, _to=501)]
            [executor.submit(self.getBalances, _data=data,  _from=501, _to=601)]
            [executor.submit(self.getBalances, _data=data,  _from=601, _to=701)]
            [executor.submit(self.getBalances, _data=data,  _from=701, _to=801)]
            [executor.submit(self.getBalances, _data=data,  _from=801, _to=901)]
            [executor.submit(self.getBalances, _data=data,  _from=901, _to=1000)]

        return data



    def getBalances(self, _data, _from, _to):
        for i in range (_from, _to):
            # == Getting Balances from each wallet == #
            get_balance = self.web3.eth.get_balance(self.wallet_address[i])

            # == Appending in self.data == #
            _data["Address"].append(self.wallet_address[i])
            _data["Balance"].append(get_balance)

            # == Print and time.sleep(100ms) == #
            print(f"Found: {self.wallet_address[i], get_balance}\n") #printing process.
            time.sleep(0.1) #change this conform to your max limit (in my test 100ms takes 40 seconds to finish.)

        return _data



    def saveCsv(self):
        #== Creating new CSV File ==#
        headers = ["Address","Balance"]
        new_df = pd.DataFrame(columns=headers)

        new_df["Address"] = self.data["Address"]
        new_df["Balance"] = self.data["Balance"]

        new_df.to_csv(r"./Ethereum/newBalances.csv", index=False) #save   



multiGetBalanceExample()

標籤：Python异步以太坊

上一篇：異步任務使應用程式崩潰

下一篇：eventlisteners回呼函式是否進入回呼佇列？