web3.py是基于python的以太坊库，内部封装了对于etereum的rpc请求，这篇文章介绍了使用web3.py v6与以太坊或以太坊系列的区块链网络进行交互的常用方式。

安装web3.py

pip install web3

加载以太坊节点：

from web3 import Web3
w3 = Web3(Web3.HTTPProvider('https://<your-provider-url>'))

如果是本地节点，例如ganache,地址为http://localhost:8545如果是线上节点，一般使用一些节点服务商提供的节点地址，例如Infura、quicknode、alchemy等。小狐狸metamast只是个钱包入口，它不提供节点api，它本身使用的也是infur提供的节点。智能合约开发框架truffle、truffle都提供了本地用于测试的私链，同样可以通过节点连接。

获取区块详情

w3.eth.get_block('latest')

这个方法可以使用'latset'获取最新块，也可以指定高度、区块哈希获取指定高度，返回一堆区块头数据

{'difficulty': 1,
 'gasLimit': 6283185,
 'gasUsed': 0,
 'hash': HexBytes('0x0'),
 'logsBloom': HexBytes('0x000'),
 'miner': '0x0',
 'mixHash': HexBytes('0x0'),
 'nonce': HexBytes('0x0'),
 'number': 0,
 'parentHash': HexBytes('0x000'),
 'proofOfAuthorityData': HexBytes('0x000'),
 'receiptsRoot': HexBytes('0x0'),
 'sha3Uncles': HexBytes('0x0'),
 'size': 622,
 'stateRoot': HexBytes('0x0'),
 'timestamp': 0,
 'totalDifficulty': 1,
 'transactions': [],
 'transactionsRoot': HexBytes('0x0'),
 'uncles': []}

常用函数和工具方法

1. 获取当前gasprice w3.eth.gase_price，经常需要使用
2. 获取当前区块高度 w3.eth.block_number
3. 获取当前链ID w3.eth.chain_id
4. 获取某账户的余额 w3.eth.get_balance(account='0x0',block_identifier=111111) 注意是eth余额，不是token的.注意第二个参数block_identifier可以指定区块高度，或者区块哈希
5. Web3.toChecksumAddress(address)：将地址（字符串类型）转换为校验和形式，以提高地址的安全性。
6. Web3.keccak(text)：计算一个字符串的 Keccak-256 哈希值。在 Solidity 中，也可以使用 keccak256 函数计算一个字符串的哈希值。
7. Web3.is_address(address)：判断一个地址是否为合法的以太坊地址。
8. Web3.to_wei(1,'ether') 数值转化,将ether转为wei
9. Web3.from_wei(1000000000000000000, 'ether'):将单位wei转化为其他单位

以太坊原生交易

受益于以太坊的常规账户模型，转账的逻辑要比BTC的UTXO模型要简洁清晰得多，在以太坊系统中从A向B转账，就是把A账户中的余额扣除，把B账户中的余额增加。以下是转账的基本流程代码：

from eth_account.account import Account, LocalAccount, SignedTransaction  
from web3 import Web3, HTTPProvider  
  
# 以太坊原生交易s  
sender_addr = "0x1111111111111111111111111111..."  
sender_private_key = "0xffffffffffffffffffffffff..."  
to_address = "0xB1476dFdAD625D787A006Ed362EBa5872a88Ae1A"  
  
w3 = Web3(HTTPProvider("http://127.0.0.1:8545"))  
## 1、构建交易数据  
tx = {  
    'from': sender_addr,  # 如果传入的地址不是发送者的地址，那么会出现typeerror的错误  
    'to': to_address,  
    'value': w3.to_wei(1, 'ether'),  # 这里的单位是wei,10**18 wei = 1 ether
    'gas': 21000,  
    'gasPrice': w3.eth.gas_price,  
    'nonce': w3.eth.get_transaction_count(sender_addr)  
}  
## 2、签名交易  
account: LocalAccount = Account.from_key(sender_private_key)  
sign_tx: SignedTransaction = account.sign_transaction(tx)  
## 3、发送交易  
tx_hash = w3.eth.send_raw_transaction(sign_tx.rawTransaction)  
## 4、阻塞等待交易结果  
result = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=60)  
print(result)

注意：

1. 交易数据中from这个字段可以不传的，但如果使用，必须保证地址与使用的私钥匹配，否则无法进行签名
2. 交易能否执行成功与gas有关，gas过低会造成以太坊虚拟机无法执行事务，高了会返回多余gas，原生交易固定为21000，其他交易可以预估web3.eth.estimate_gas(tx)，但一般直接上限写死即可，比如100000
3. 能否上链被矿工打包与gas无关，与gasPrice相关，gas价格太低了矿工嫌弃你，太高了你嫌弃矿工，web3.py提供了预估当前最佳gasprice的方法w3.eth.gase_price，你和矿工都满意
4. value这个字段是交易的值，单位是Wei，对于原生交易，这个值表示从A转给B的钱，对于合约交易，这个值一般为0。
5. nonce这个值是记录交易发起者交易序列的，每次转账都比上一次的nonce加1，使用wb3.eth.get_transaction_count(sender)获取sender的nonce值(注意，这个值已经进行了加1)
6. web3.py的签名过程不同版本还不一样，这里使用的是v6.1.0
7. 这里使用的是eth_account提供的Account进行签名，对于原生交易和合约交易效果一样，但可能对于一些新EIP的交易类型不支持
8. 签名的两个参数，一个是需要签名的交易数据，一个是私钥，注意私钥的保密
9. 签名完成后会生成Raw交易数据，结构如下，实际上最后用到的只有rawTransaction这个字段

{
 'hash': HexBytes('0x126431f2a7fda003aada7c2ce52b0ce3cbdbb1896230d3333b9eea24f42d15b0'), 
 'r': 110093478023675319011132687961420618950720745285952062287904334878381994888509,
 'rawTransaction':HexBytes('0x........'), 
 's': 33674551144139401179914073499472892825822542092106065756005379322302694600392,  
 'v': 0}

注意，发送过程是异步的，以太坊从交易发送到节点确认需要一定的时间，主网一般10～15秒，goerli这类测试网要快一点，如arbitrum这类L2的网络最快基本上1～3秒。
发送交易后可以等待交易确认再执行后续操作，使用方法：

resp = w3.eth.wait_for_transaction_receipt(tx_hash)

这个方法其实是一个死循环在请求节点获取交易哈希状态结果，默认超时120秒

可以将上述流程封装成函数，方便调用

from eth_account.account import Account, LocalAccount, SignedTransaction  
from hexbytes import HexBytes  
from web3 import Web3, HTTPProvider  
  
  
def transfer(w3: Web3, account: LocalAccount, to: str, value: int, gas=21000) -> HexBytes:  
     """
    交易、普通转账
    :param w3: 以太坊客户端Web3实例
    :param account: 以太坊账户，由eth-account提供
    :param to: 接收地址
    :param value: 转账金额,单位时wei,1 ether=10**18 wei
    :param gas: gas消耗，默认21000，不能比这个低
    :return: 交易哈希 字节数组
    """
    ## 1、构建交易数据  
    tx = {  
        'from': account.address,  
        'to': to,  
        'value': value,  
        'gas': gas,  
        'gasPrice': w3.eth.gas_price,  
        'nonce': w3.eth.get_transaction_count(account.address)  
    }  
    ## 2、签名  
    sign_tx: SignedTransaction = account.sign_transaction(tx)  
    ## 3、发送  
    return w3.eth.send_raw_transaction(sign_tx.rawTransaction)
    
if __name__ == '__main__':  
    w3 = Web3(HTTPProvider(endpoint_uri='http://127.0.0.1:8545'))  
    account = Account.from_key("<你的私钥>")  
    tx_hash = transfer(w3, account, '<接收地址>', 1*10**17)  
    print(f"Success Transfer Hash:{tx_hash.hex()}")  
    result = w3.eth.wait_for_transaction_receipt(tx_hash)  
    print(f"Transfer Receipt:{result}")

调用合约方法

智能合约一旦部署就不允许再修改，后续对于合约的使用绝大部分集中在合约方法调用和过滤合约日志

1. 智能合约的方法从外部调用的角度可以分为两种，一种是view方法，一种是非view方法。

• view方法只是从区块链上查询数据，不会进行数据更改，不需要调用者进行签名操作
• 非view方法需要修改区块链上数据，需要调用者进行签名操作

2. 调用合约需要生成一个合约实例，这取决于两个东西，一是合约address，二是合约ABI

在本文中，需要部署一个简单的合约到区块链上:

// SPDX-License-Identifier: GPL-3.0
pragma solidity ^0.8.9;

contract Demo {
    mapping(address => uint) private _ages;

    function setAge(uint age) public {
        _ages[msg.sender] = age;
    }
    function getAge(address user) public view returns (uint) {
        return _ages[user];
    }
}

在上面的合约中，提供了一个状态变量_ages来记录地址和年龄的映射关系，并提供了两个方法，setAge(uint age)设置调用者的年龄，getAge(address user)获取指定地址的年龄。

调用合约的View方法:

对于View方法，调用方法分为几步：

1. 加载Web3实例
2. 加载合约实例
3. 获取合约方法对象
4. 使用call()方法调用

from web3 import Web3, HTTPProvider

# 连接以太坊节点
w3 = Web3(HTTPProvider(endpoint_uri='http://127.0.0.1:8545'))

# 获取合约实例
contract_address = '<合约地址>'
contract_abi = [] # 合约abi,这是一个list结构
contract_instance = w3.eth.contract(address=contract_address, abi=contract_abi)

# 获取View函数对象并调用
func = getattr(contract_instance.functions, 'getAge')(user='0xC3845C061D0e0929744B01CbB6c23c31402B3E3a')
result=func.call()
print(f"result={result})

调用合约非View方法

对于非View方法，则分为下面几步：

1. 加载Web3实例
2. 加载合约实例
3. 获取合约方法对象
4. 将方法对象绑定到交易数据中的data字段中
5. 对交易数据进行签名
6. 发送签名后的tx

from eth_account.account import Account, SignedTransaction
from web3 import Web3, HTTPProvider

# 调用者地址
sender_address = '<调用者地址>'
sender_privatekey = '<调用者私钥>'

# 连接以太坊节点
w3 = Web3(HTTPProvider(endpoint_uri='http://127.0.0.1:8545'))

# 获取合约实例
contract_address = '0x6EC21D47FCF6F06eeE9b36Ee85Ae25417b44697a'
contract_abi = [] # 合约ABI，这是一个List结构
contract_instance = w3.eth.contract(address=contract_address, abi=contract_abi)

# 获取调用函数对象
func = getattr(contract_instance.functions, 'setAge')(age=18)

# 构造交易数据
tx = func.build_transaction({
    'nonce': w3.eth.get_transaction_count(sender_address),
    'value': 0,
    'gas': 100000,
    'gasPrice': w3.eth.gas_price})

sign_tx: SignedTransaction = Account.sign_transaction(tx, sender_privatekey)
tx_hash = w3.eth.send_raw_transaction(sign_tx.rawTransaction)
print(tx_hash.hex())

同样，可以将上述过程封装成函数，方便调用：

from eth_account.account import Account, LocalAccount, SignedTransaction
from web3 import Web3, HTTPProvider


def call_contract_sign_func(w3: Web3, account: LocalAccount, address: str, abi: list, func_name: str,
                            gas=200000, value=0, **kwargs):
    """
    调用合约非View方法
    :param w3: Web3 实例
    :param account: 账户实例
    :param address: 合约地址
    :param abi: 合约ABI
    :param func_name: 合约方法名
    :param gas: 燃气,默认200000
    :param value: 附带转账金额，默认0
    :param kwargs: 方法参数
    :return: 交易哈希字节数组
    """
    contract_instance = w3.eth.contract(address=address, abi=abi)
    func = getattr(contract_instance.functions, func_name)(**kwargs)
    # 构造交易数据
    tx = func.build_transaction({
        'nonce': w3.eth.get_transaction_count(account.address),
        'value': value,
        'gas': gas,
        'gasPrice': w3.eth.gas_price})
    # 签名
    sign_tx: SignedTransaction = account.sign_transaction(tx)
    # 发送
    return w3.eth.send_raw_transaction(sign_tx.rawTransaction)


def call_contract_view_func(w3, address, abi, func_name, **kwargs):
    """
    调用合约View方法
    :param w3:Web3实例
    :param address:合约地址
    :param abi:合约ABI
    :param func_name:合约方法名称
    :param kwargs:合约方法参数
    :return:函数执行结果
    """
    contract_instance = w3.eth.contract(address=address, abi=abi)
    func = getattr(contract_instance.functions, func_name)(**kwargs)
    return func.call()


if __name__ == '__main__':
    w = Web3(HTTPProvider(endpoint_uri='http://127.0.0.1:8545'))
    contract_address = '<合约地址>'
    contract_abi = [] # 合约abi
    caller: LocalAccount = Account.from_key('<账户私钥>')
    ## 调用view方法
    result = call_contract_view_func(w3=w, address=contract_address, abi=contract_abi, func_name='getAge',
                                     user='0xC3845C061D0e0929744B01CbB6c23c31402B3E3a')
    print(result)
        ## 调用非View方法
    tx_hash = call_contract_sign_func(w3=w, account=caller, address=contract_address, abi=contract_abi,
                                      func_name='setAge',
                                      age=8877)
    print(tx_hash.hex())
    ## 等待并返回调用结果
    resp = w.eth.wait_for_transaction_receipt(tx_hash)
    print(resp)

上面的代码中可以看出调用合约非view方法实际上就是一次转账交易，在交易中把对合约方法的调用信息写入交易数据的data字段中，剩下的就是evm需要考虑的事情了。

过滤合约日志

日志的过滤是合约使用的另一个重要方面，为了便于日志演示，对上面使用的合约进行简单的改造，加上event：

// SPDX-License-Identifier: GPL-3.0
pragma solidity ^0.8.9;

contract Demo {
    // 日志
    event SetAge(address user, uint age);
    mapping(address => uint) private _ages;
    function setAge(uint age) public {
        _ages[msg.sender] = age;
      // 发射日志
        emit SetAge(msg.sender, age);
    }

    function getAge(address user) public view returns (uint) {
        return _ages[user];
    }
}

过滤合约日志分如下几步：

1. 加载web3实例
2. 加载合约实例
3. 获取合约事件对象
4. 创建事件过滤器
5. 检索过滤后的全部事件

from web3 import Web3, HTTPProvider

contract_address = '<合约地址>'
contract_abi = [] # 合约ABI
# 加载Web3实例
w3 = Web3(HTTPProvider(endpoint_uri='http://127.0.0.1:8545'))
# 加载合约实例
contract_instance = w3.eth.contract(address=contract_address, abi=contract_abi)
# 获取event对象
my_event = contract_instance.events.SetAge()
# 创建事件过滤器，
my_filter = my_event.create_filter(fromBlock=1, toBlock=28)
# 检索事件
for event in my_filter.get_all_entries():
    print(dict(event))
   

返回的event数据如下，包含了日志的基本信息：

• 区块信息：区块高度、哈希
• 交易哈希、交易哈希索引
• 日志索引
• event名
• event 数据信息

{
 'args':AttributeDict({'user': '0xC3845C061D0e0929744B01CbB6c23c31402B3E3a', 'age': 8877}), 
 'event': 'SetAge',
 'logIndex': 0,
 'transactionIndex': 0,
 'transactionHash': HexBytes('0x0975e324106b995512a331f0fa29e8af1ef607e41753c84cacf5a6009dda34e7'), 
 'address': '0x9d95b127281335A846E4f36e36D45963e452d3D4',
 'blockHash': HexBytes('0x1b67007d4f40c93c1e881a32299541cffe837b8acc64724ac093ae7d9bc4e832'),
 'blockNumber': 28
}

上面的代码中并没有体现log的topic信息，这里需要首先说明 Log 、event、topic的关系，以太坊的Log数据结构如下：

```go
type Log struct {  
   // 日志所属的区块高度  
   BlockNumber uint64 `json:"blockNumber"`  
   // 日志所属的交易哈希  
   BlockHash common.Hash `json:"blockHash"`  
   // 日志所属的区块哈希  
   TxHash common.Hash `json:"transactionHash" gencodec:"required"`  
   // 日志所属的交易序列号  
   TxIndex uint `json:"transactionIndex"`  
  
   // 日志在交易中的序列，一个交易可以包含多个日志，这个字段表示日志在交易中的索引  
   Index uint `json:"logIndex"`  
   // 是否被移除，注意如果链进行了分叉，那么本字段会变成true，如果为true，那么不应该识别为正确的log  
   Removed bool `json:"removed"`  
  
   // 发送出日志的智能合约地址，注意，只有合约账户才能发送event  
   Address common.Address `json:"address" gencodec:"required"`  
   // 日志主题列表，这个字段是为了方便查询时进行过滤  
   Topics []common.Hash `json:"topics" gencodec:"required"`  
   // event携带的数据字节数组  
   Data []byte `json:"data" gencodec:"required"`  
}
}
```

Log 是以太坊中的一个事件记录，是一个 key-value 对的集合，其中 key 是字符串类型的 topics，value 是一个任意长度的字节数组 data。Log 可以被合约 emit，也可以由以太坊系统自动生成，例如在转账时生成的 Transfer Log。每笔交易都有可能会产生 Log。

Event 是合约中的一类特殊函数，它允许合约在执行过程中，向外部环境广播某些内容（包括 Log 和数据）。Event 可以定义多个参数，这些参数可以是基本类型（如 uint256、string 等），也可以是自定义类型。Event 可以被触发多次，并且每次触发都会生成一个独立的 Log。

Topic 是 Log 中的一个结构。在 Emit Event 的时候，每个 Event 参数可能会被标记为 indexed 或不 indexed，indexed 的参数会被记录在 Event 的 Topic 中，不 indexed 的参数则会被记录在 Log 的 Data 中。具体来说，Event 的 keccak 哈希会成为 Topic 的第一个元素，Events 中 indexed 的参数会按照顺序在后面添加。

上面过滤日志的代码中，只是通过合约地址和区块范围进行过滤，如果要过滤event和或者过滤方法参数那么需要在构造过滤器时添加topics字段，

每个log都有一个topic集合，每个event至少有一个topic，即event方法的签名

topic=w3.keccak(text="SetAge(address,uint256)").hex()

如果event的参数被indexed修饰，那么也可以生成topic

topic=w3.keccak(text="user(address)").hex()

在构建过滤器时添加topics

my_filter = my_event.create_filter(fromBlock=1, toBlock=28, topics=[w3.keccak(text="SetAge(address,uint256)").hex()])

topic相当于给日志打标签，通过标签可以提高过滤效率，但每次要计算哈希还是挺麻烦的，如果logs数量本身不大，可以直接全量过滤后再通过event name。

Web3.py 还提供了很多其他有用的工具，比如用于解析和生成合约 ABI 编码的方法。但合约的部署、测试一般由hardhat、truffle这类框架完成，使用的是js或者ts,这方面python的存在感比较低。

参考：

1. web3.py官网文档