如何在Python类中正确调用异步函数？

错误分析 #

你的代码存在两个核心问题：

1. 静态方法错误使用self参数：@staticmethod装饰的方法不需要self参数，因为静态方法属于类本身，不绑定实例，传入self会导致调用时参数不匹配。
2. 异步方法未通过asyncio启动：类中的异步方法和普通异步函数一样，必须通过asyncio.run()来启动事件循环，直接调用只会返回一个协程对象，不会执行。

修正后的代码 #

import json
import asyncio
import httpx
from web3 import Web3

url_1inch_bsc = 'https://api.1inch.io/v5.0/56/quote/'
web3_bsc = Web3(Web3.HTTPProvider('https://bscrpc.com'))

class Pars:
    @staticmethod
    async def parsing_bsc_1inch():
        # 改用with语句自动关闭文件，更安全
        with open('pairs_bsc.txt', 'r') as bsc_pairs:
            for line in bsc_pairs:
                line = line.rstrip().split('&')
                params={
                    'fromTokenAddress': line[1], 
                    'toTokenAddress': line[2], 
                    'amount': '1000000000000000000'
                }
                async with httpx.AsyncClient() as client:
                    try: 
                        response = await client.get(url=url_1inch_bsc, params=params)
                        # 用response.json()直接解析，更简洁
                        json_response = response.json()
                        print(f'{line[0]} {web3_bsc.fromWei(int(json_response["fromTokenAmount"]), "ether")} -> {web3_bsc.fromWei(int(json_response["toTokenAmount"]), "ether")}')
                    except Exception as e:
                        # 捕获具体异常，方便排查错误
                        print(f'Error with request to bsc 1inch: {str(e)}') 

if __name__ == "__main__":
    asyncio.run(Pars.parsing_bsc_1inch())

关键修改说明 #

• 移除了静态方法中的self参数，静态方法不需要实例引用，直接通过类名.方法名调用即可。
• 用with语句管理文件，避免手动关闭可能出现的遗漏问题。
• 替换json.loads(response.text)为response.json()，是httpx提供的更便捷的JSON解析方式。
• 捕获具体的Exception并打印异常信息，比空except更容易定位问题。
• 启动方式改为asyncio.run(Pars.parsing_bsc_1inch())，直接调用类的静态异步方法，由asyncio启动事件循环执行。

可选方案：用实例方法实现 #

如果你不需要静态方法，也可以改成实例方法，此时需要保留self参数，调用时先实例化类，再传入asyncio.run()：

class Pars:
    async def parsing_bsc_1inch(self):
        # 方法内容和上面一致，保留self
        with open('pairs_bsc.txt', 'r') as bsc_pairs:
            # ... 其余代码不变

if __name__ == "__main__":
    pars_instance = Pars()
    asyncio.run(pars_instance.parsing_bsc_1inch())

内容的提问来源于stack exchange，提问作者stupidnewbeeee