如何使用 web3.py 获取合约地址的待处理交易？答案

【问题标题】：How can I get pending transaction of contract address using web3.py?如何使用 web3.py 获取合约地址的待处理交易？
【发布时间】：2022-10-06 15:27:04
【问题描述】：

我想得到一个合约地址的待处理交易，我尝试了很多方法但没有奏效

方法1：这似乎擅长对待处理的交易进行排序，但我无法从我的地址获得任何交易，我不知道为什么。请帮我

def main():
    block_filter = web3.eth.filter(\'pending\') 
    log_loop(block_filter, 0)

def log_loop(block_filter, poll_interval):
    while True: 
        for event in block_filter.get_new_entries():
            if web3.eth.getTransaction(event)[\'from\'] == my contract:
                print(event)

方法 2：这有助于我从我的地址获取交易，但它获得的所有交易都已确认，而不是待处理

def main():
    block_filter = web3.eth.filter({\'fromBlock\':\'pending\',\'toBlock\':\'pending\', \'address\':contract_address}) #this is not working, return nothing

    #block_filter = web3.eth.filter({\'fromBlock\':0,\'toBlock\':\'pending\', \'address\':contract_address}) #return confirmed transaction, not pending

    #block_filter = web3.eth.filter({\'fromBlock\':\'pending\',\'toBlock\':\'latest\', \'address\':contract_address}) #return confirmed transaction, not pending

    #block_filter = web3.eth.filter({\'fromBlock\':\'latest\',\'toBlock\':\'pending\', \'address\':contract_address}) #return error from > to

    #block_filter = web3.eth.filter({\'address\':contract_address}) #return confirmed transaction, not pending
    log_loop(block_filter, 0)

def log_loop(block_filter, poll_interval):
    while True: 
        for event in block_filter.get_new_entries():
            print(event)

标签： python transactions token web3py pending-transition

【解决方案1】：

您可以订阅新的待处理事务，使用 WebSocket 连接到您的节点有关更多信息，请查看https://geth.ethereum.org/docs/rpc/pubsub

您将收到所有待处理的交易 txHash。之后，您可以过滤它们并仅处理所需的令牌地址。一种方法是对每个接收到的 txHash 使用 web3.eth.get_transaction()。它将返回一个包含交易详细信息的字典使用此字典中的“to”键来过滤是否已发送到所需合约地址的待处理交易。贝娄是（稍作修改的）解决方案，建议在： https://community.infura.io/t/web3-py-how-to-subscribe-to-pending-ethereum-transactions-in-python/5409

import asyncio
import json
import requests
import websockets
from web3 import Web3

web3 = Web3(Web3.HTTPProvider(node_http_url))
# specific address to watch at 
watched_address = '<CHOOSED_PUBLIC_ADDRESS>' 

async def get_event():
    async with websockets.connect(node_ws_url) as ws:
        await ws.send('{"jsonrpc": "2.0", "id": 1, "method": "eth_subscribe", "params": ["newPendingTransactions"]}')
        subscription_response = await ws.recv()
        print(subscription_response)

        while True:
            try:
                message = await asyncio.wait_for(ws.recv(), timeout=15)
                response = json.loads(message)
                txHash = response['params']['result']
                print(txHash)

                tx = web3.eth.get_transaction(txHash)
                if tx.to == watched_address:
                    print("Pending transaction found with the following details:")
                    print({
                        "hash": txHash,
                        "from": tx["from"],
                        "value": web3.fromWei(tx["value"], 'ether')
                    })
                pass
            except:
                pass

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    while True:
        loop.run_until_complete(get_event())

【讨论】：