如何在Python循环中使用异步处理错误？

Question

我正在使用 Etherscan API 收集从许多地址发送的最后一次交易的日期。我做了 python 循环来做到这一点。您可以在下面看到代码

from datetime import datetime
from etherscan import Etherscan
import csv
import sys
import time

# etherscan api
eth = Etherscan("YOUR_ETHERSCAN_API") # key in quotation marks

# import csv
with open('input/wallets.csv', 'r') as file:
    next(file) # drop header
    wallets = [row[0] for row in csv.reader(file, delimiter=",")]

# loop
for walletaddress in wallets:
    # parse json
    my_list = eth.get_normal_txs_by_address(address=(walletaddress), startblock="8200000", endblock="18998125", sort="srt")
    des_ind = my_list[-1]
    # convert timestamp to date
    ts = int(des_ind['timeStamp'])
    print(walletaddress + ',' + datetime.utcfromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S'))
    print(walletaddress + ',' + datetime.utcfromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S'), file=open(('output/output.csv'), mode='a', encoding="utf-8"))
    time.sleep(0.2) # Sleep for 0.2 seconds

一切正常，直到我面对一个没有发送交易的钱包。我会得到下面的错误

$ python3 main.py
0xfbb1b73c4f0bda4f67dca266ce6ef42f520fbb98,2019-07-25 00:55:39
0x201f981aa5fcf15e5c8305448614b5e5fb9f195d,2023-12-20 05:46:23
0x5094ec3773ac3122b70e4c93295f8126862ff71d,2024-01-11 19:11:11
Traceback (most recent call last):
  File "/home/debian/scrape/ETH/main.py", line 18, in <module>
    my_list = eth.get_normal_txs_by_address(address=(walletaddress), startblock="8200000", endblock="18998125", sort="srt")
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/debian/.local/lib/python3.11/site-packages/etherscan/etherscan.py", line 33, in wrapper
    return parser.parse(r)
           ^^^^^^^^^^^^^^^
  File "/home/debian/.local/lib/python3.11/site-packages/etherscan/utils/parsing.py", line 12, in parse
    assert status, f"{result} -- {message}"
AssertionError: [] -- No transactions found

我在某处读到可以使用异步来处理错误，因为我不希望循环中断。我尝试了以下几行代码

async with Etherscan("YOUR_ETHERSCAN_API") as eth:
for walletaddress in wallets:
return await eth.get_normal_txs_by_address(address=(walletaddress), startblock="8200000", endblock="18998125", sort="srt")
except Exception as E:
if str(E) == '[] -- No transactions found':
return []
else:
print(f'Cannot request history from {startblock} to {endblock} due to {E}')

我几乎一整天都在绕着它转。我无法让它与我的代码一起工作。我看到的几个答案与我的循环不同。我认为这个问题的答案将让其他人了解如何处理类似循环中的错误。

如何处理上述错误，以便我的循环继续并且不会中断？希望有人能帮忙

Answer 1

我尝试在循环内使用

try

和

except

并添加

aiohttp

功能，并且不再出现错误。看看它是否也适合您：

pip install aiohttp

对代码进行一些调整，例如：

import asyncio
from datetime import datetime
import aiohttp
import csv

async def fetch_transactions(session, eth, walletaddress):
    try:
        async with session.get(
            f"https://api.etherscan.io/api",
            params={
                "module": "account",
                "action": "txlist",
                "address": walletaddress,
                "startblock": "8200000",
                "endblock": "18998125",
                "sort": "asc",
                "apikey": eth.api_key,
            },
        ) as response:
            data = await response.json()
            my_list = data.get("result", [])
            des_ind = my_list[-1]
            ts = int(des_ind['timeStamp'])
            print(walletaddress + ',' + datetime.utcfromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S'))
            print(walletaddress + ',' + datetime.utcfromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S'), file=open(('output/output.csv'), mode='a', encoding="utf-8"))
            return True
    except Exception as e:
        if str(e) == '[]':
            print(f"No transactions found for {walletaddress}")
            return False
        else:
            print(f'Cannot request history for {walletaddress} due to {e}')
            return False

async def main():
    # etherscan api
    class Etherscan:
        def __init__(self, api_key):
            self.api_key = api_key

    eth = Etherscan("YOUR_ETHERSCAN_API")

    # import csv
    with open('input/wallets.csv', 'r') as file:
        next(file, None)  # drop header
        wallets = [row[0] for row in csv.reader(file, delimiter=",")]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_transactions(session, eth, walletaddress) for walletaddress in wallets]
        results = await asyncio.gather(*tasks)

        # Handle results as needed
        print("Script completed. Results:", results)

if __name__ == "__main__":
    asyncio.run(main())

当然，由于我得到的测试环境为空：

Script completed. Results: []

但是，在具有真实数据集的完整软件上，如果匹配，应该返回一些数据。