Python与Web3.py库交互实践

Python与Web3.py库交互实践

引言：连接Python与区块链的桥梁

Web3.py是以太坊官方推荐的Python接口库，它使开发者能够通过Python与以太坊区块链进行交互。本指南将带你从基础到高级应用，全面掌握使用Web3.py进行区块链交互的核心技能，包括智能合约部署、交易签名、事件监听等关键功能。

1. 环境配置与基础连接

1.1 安装Web3.py

pip install web3

1.2 连接以太坊节点

from web3 import Web3# 使用Infura连接主网
infura_url = "https://mainnet.infura.io/v3/YOUR_INFURA_PROJECT_ID"
w3 = Web3(Web3.HTTPProvider(infura_url))# 检查连接
if w3.is_connected():print("Connected to Ethereum Mainnet")print(f"Latest block: {w3.eth.block_number}")
else:print("Connection failed")# 连接本地开发节点（如Ganache）
ganache_url = "http://127.0.0.1:7545"
w3_ganache = Web3(Web3.HTTPProvider(ganache_url))

2. 基础区块链交互

2.1 账户与余额查询

# 查询账户余额（单位：wei）
balance_wei = w3.eth.get_balance("0x742d35Cc6634C0532925a3b844Bc454e4438f44e")# 转换单位
balance_eth = w3.from_wei(balance_wei, 'ether')
print(f"Balance: {balance_eth} ETH")# 获取账户列表（Ganache）
ganache_accounts = w3_ganache.eth.accounts
print(f"Ganache accounts: {ganache_accounts}")

2.2 创建并发送交易

def send_eth_transaction(w3, sender, receiver, amount_eth, private_key):"""发送ETH交易"""# 转换金额单位amount_wei = w3.to_wei(amount_eth, 'ether')# 获取noncenonce = w3.eth.get_transaction_count(sender)# 构建交易字典tx = {'nonce': nonce,'to': receiver,'value': amount_wei,'gas': 21000,  # 标准ETH转账的gas limit'gasPrice': w3.to_wei('50', 'gwei'),'chainId': w3.eth.chain_id}# 签名交易signed_tx = w3.eth.account.sign_transaction(tx, private_key)# 发送交易tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)# 等待交易确认receipt = w3.eth.wait_for_transaction_receipt(tx_hash)return receipt# 在Ganache上测试
sender = ganache_accounts[0]
receiver = ganache_accounts[1]
private_key = "0x..."  # 替换为Ganache中第一个账户的私钥receipt = send_eth_transaction(w3_ganache, sender, receiver, 1.0, private_key)
print(f"Transaction receipt: {receipt}")

3. 智能合约交互

3.1 加载和部署合约

from solcx import compile_source# 编译合约
contract_source_code = """
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;contract SimpleStorage {uint256 storedData;function set(uint256 x) public {storedData = x;}function get() public view returns (uint256) {return storedData;}
}
"""# 编译合约
compiled_sol = compile_source(contract_source_code)
contract_interface = compiled_sol['<stdin>:SimpleStorage']# 部署合约
def deploy_contract(w3, contract_interface, deployer_private_key):# 获取部署者账户deployer = w3.eth.account.from_key(deployer_private_key).address# 创建合约对象Contract = w3.eth.contract(abi=contract_interface['abi'],bytecode=contract_interface['bin'])# 构建部署交易tx = Contract.constructor().build_transaction({'from': deployer,'nonce': w3.eth.get_transaction_count(deployer),'gas': 2000000,'gasPrice': w3.to_wei('50', 'gwei')})# 签名并发送signed_tx = w3.eth.account.sign_transaction(tx, deployer_private_key)tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)# 等待部署完成receipt = w3.eth.wait_for_transaction_receipt(tx_hash)return receipt.contractAddress# 在Ganache上部署
contract_address = deploy_contract(w3_ganache, contract_interface, private_key)
print(f"Contract deployed at: {contract_address}")

3.2 与已部署合约交互

# 创建合约实例
contract = w3_ganache.eth.contract(address=contract_address,abi=contract_interface['abi']
)# 调用只读函数
current_value = contract.functions.get().call()
print(f"Current stored value: {current_value}")# 调用状态变更函数
def set_contract_value(w3, contract, new_value, caller_private_key):caller = w3.eth.account.from_key(caller_private_key).address# 构建交易tx = contract.functions.set(new_value).build_transaction({'from': caller,'nonce': w3.eth.get_transaction_count(caller),'gas': 2000000,'gasPrice': w3.to_wei('50', 'gwei')})# 签名并发送signed_tx = w3.eth.account.sign_transaction(tx, caller_private_key)tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)# 等待交易确认receipt = w3.eth.wait_for_transaction_receipt(tx_hash)return receipt# 更新合约状态
receipt = set_contract_value(w3_ganache, contract, 42, private_key)
print(f"Set value transaction receipt: {receipt}")# 验证更新
updated_value = contract.functions.get().call()
print(f"Updated stored value: {updated_value}")

4. 高级功能实践

4.1 事件监听

# 增强合约以包含事件
contract_with_event = """
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;contract EventfulStorage {uint256 storedData;event ValueChanged(address indexed author, uint256 newValue);function set(uint256 x) public {storedData = x;emit ValueChanged(msg.sender, x);}function get() public view returns (uint256) {return storedData;}
}
"""# 部署增强版合约
compiled_event = compile_source(contract_with_event)
event_contract_interface = compiled_event['<stdin>:EventfulStorage']
event_contract_address = deploy_contract(w3_ganache, event_contract_interface, private_key)
event_contract = w3_ganache.eth.contract(address=event_contract_address,abi=event_contract_interface['abi']
)# 创建事件过滤器
value_changed_filter = event_contract.events.ValueChanged.create_filter(fromBlock='latest'
)# 监听事件
def event_listener():print("Listening for ValueChanged events...")while True:for event in value_changed_filter.get_new_entries():print(f"New event: Author={event.args.author}, Value={event.args.newValue}")time.sleep(2)# 在后台线程运行监听器
import threading
thread = threading.Thread(target=event_listener, daemon=True)
thread.start()# 触发事件（在另一个线程）
def trigger_events():for i in range(5):set_contract_value(w3_ganache, event_contract, i, private_key)time.sleep(3)trigger_thread = threading.Thread(target=trigger_events)
trigger_thread.start()

4.2 与ERC-20代币交互

# ERC-20代币ABI（简化版）
erc20_abi = [{"name": "balanceOf","outputs": [{"type": "uint256", "name": ""}],"inputs": [{"type": "address", "name": "account"}],"stateMutability": "view","type": "function"},{"name": "transfer","outputs": [{"type": "bool", "name": ""}],"inputs": [{"type": "address", "name": "recipient"},{"type": "uint256", "name": "amount"}],"stateMutability": "nonpayable","type": "function"}
]# 连接DAI合约
dai_address = "0x6B175474E89094C44Da98b954EedeAC495271d0F"  # 主网DAI合约
dai_contract = w3.eth.contract(address=dai_address, abi=erc20_abi)# 查询余额
wallet_address = "0x742d35Cc6634C0532925a3b844Bc454e4438f44e"
balance = dai_contract.functions.balanceOf(wallet_address).call()
print(f"DAI balance: {balance / (10 ** 18)} DAI")# 转账函数（需要私钥）
def transfer_erc20(w3, contract, recipient, amount, sender_private_key):sender = w3.eth.account.from_key(sender_private_key).address# 构建交易tx = contract.functions.transfer(recipient,amount).build_transaction({'from': sender,'nonce': w3.eth.get_transaction_count(sender),'gas': 100000,'gasPrice': w3.to_wei('50', 'gwei'),'chainId': w3.eth.chain_id})# 签名并发送signed_tx = w3.eth.account.sign_transaction(tx, sender_private_key)tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)# 等待确认receipt = w3.eth.wait_for_transaction_receipt(tx_hash)return receipt# 注意：实际主网操作需要真实资金和谨慎处理私钥

5. 完整DApp示例：链上记事本

5.1 智能合约

// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;contract OnchainNotepad {struct Note {string content;uint256 timestamp;}mapping(address => Note[]) private userNotes;event NoteAdded(address indexed user, uint256 index, string content);event NoteUpdated(address indexed user, uint256 index, string content);function addNote(string memory content) public {userNotes[msg.sender].push(Note(content, block.timestamp));emit NoteAdded(msg.sender, userNotes[msg.sender].length - 1, content);}function updateNote(uint256 index, string memory content) public {require(index < userNotes[msg.sender].length, "Note does not exist");userNotes[msg.sender][index].content = content;userNotes[msg.sender][index].timestamp = block.timestamp;emit NoteUpdated(msg.sender, index, content);}function getNoteCount() public view returns (uint256) {return userNotes[msg.sender].length;}function getNote(uint256 index) public view returns (string memory content, uint256 timestamp) {require(index < userNotes[msg.sender].length, "Note does not exist");Note storage note = userNotes[msg.sender][index];return (note.content, note.timestamp);}
}

5.2 Python后端

from flask import Flask, jsonify, request
from web3 import Web3
import jsonapp = Flask(__name__)
w3 = Web3(Web3.HTTPProvider("http://127.0.0.1:7545"))# 加载合约
with open('OnchainNotepad.json') as f:contract_data = json.load(f)contract_address = "0xYourDeployedContractAddress"
contract = w3.eth.contract(address=contract_address, abi=contract_data['abi'])@app.route('/notes', methods=['GET'])
def get_notes():address = request.args.get('address')if not address or not w3.is_address(address):return jsonify({"error": "Invalid address"}), 400note_count = contract.functions.getNoteCount().call({'from': address})notes = []for i in range(note_count):content, timestamp = contract.functions.getNote(i).call({'from': address})notes.append({"index": i,"content": content,"timestamp": timestamp})return jsonify(notes)@app.route('/add', methods=['POST'])
def add_note():data = request.jsonaddress = data.get('address')content = data.get('content')private_key = data.get('private_key')if not all([address, content, private_key]):return jsonify({"error": "Missing parameters"}), 400# 构建交易tx = contract.functions.addNote(content).build_transaction({'from': address,'nonce': w3.eth.get_transaction_count(address),'gas': 300000,'gasPrice': w3.to_wei('50', 'gwei')})# 签名并发送signed_tx = w3.eth.account.sign_transaction(tx, private_key)tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)return jsonify({"tx_hash": tx_hash.hex()})if __name__ == '__main__':app.run(port=5000)

5.3 前端界面

import tkinter as tk
from web3 import Web3
import requestsclass NotepadApp:def __init__(self, root):self.root = rootself.root.title("链上记事本")# 配置self.w3 = Web3(Web3.HTTPProvider("http://127.0.0.1:7545"))self.api_url = "http://localhost:5000"# UI组件self.address_label = tk.Label(root, text="你的地址:")self.address_entry = tk.Entry(root, width=50)self.private_key_label = tk.Label(root, text="私钥(仅本地):")self.private_key_entry = tk.Entry(root, width=50, show="*")self.note_label = tk.Label(root, text="笔记内容:")self.note_text = tk.Text(root, height=10, width=50)self.add_button = tk.Button(root, text="添加笔记", command=self.add_note)self.load_button = tk.Button(root, text="加载笔记", command=self.load_notes)self.notes_listbox = tk.Listbox(root, width=70, height=15)self.notes_listbox.bind('<<ListboxSelect>>', self.on_note_select)# 布局self.address_label.grid(row=0, column=0, sticky="w")self.address_entry.grid(row=0, column=1)self.private_key_label.grid(row=1, column=0, sticky="w")self.private_key_entry.grid(row=1, column=1)self.note_label.grid(row=2, column=0, sticky="nw")self.note_text.grid(row=2, column=1)self.add_button.grid(row=3, column=1, sticky="e", pady=5)self.load_button.grid(row=4, column=1, sticky="e", pady=5)self.notes_listbox.grid(row=5, column=0, columnspan=2, padx=10, pady=10)# 状态变量self.current_notes = []self.selected_index = Nonedef add_note(self):address = self.address_entry.get()private_key = self.private_key_entry.get()content = self.note_text.get("1.0", "end-1c")if not address or not private_key or not content:tk.messagebox.showerror("错误", "请填写所有字段")returnpayload = {"address": address,"content": content,"private_key": private_key}try:response = requests.post(f"{self.api_url}/add", json=payload)if response.status_code == 200:tk.messagebox.showinfo("成功", f"笔记已添加!\n交易哈希: {response.json()['tx_hash']}")self.note_text.delete("1.0", tk.END)else:tk.messagebox.showerror("错误", f"添加失败: {response.text}")except Exception as e:tk.messagebox.showerror("错误", f"连接失败: {str(e)}")def load_notes(self):address = self.address_entry.get()if not address:tk.messagebox.showerror("错误", "请输入地址")returntry:response = requests.get(f"{self.api_url}/notes?address={address}")if response.status_code == 200:self.current_notes = response.json()self.notes_listbox.delete(0, tk.END)for i, note in enumerate(self.current_notes):preview = note['content'][:50] + "..." if len(note['content']) > 50 else note['content']timestamp = note['timestamp']self.notes_listbox.insert(tk.END, f"[{i}] {preview} (时间戳: {timestamp})")else:tk.messagebox.showerror("错误", f"加载失败: {response.text}")except Exception as e:tk.messagebox.showerror("错误", f"连接失败: {str(e)}")def on_note_select(self, event):selection = event.widget.curselection()if selection:self.selected_index = selection[0]note = self.current_notes[self.selected_index]self.note_text.delete("1.0", tk.END)self.note_text.insert(tk.END, note['content'])if __name__ == "__main__":root = tk.Tk()app = NotepadApp(root)root.mainloop()

6. 安全最佳实践

6.1 私钥管理

import os
from dotenv import load_dotenv
from web3 import Web3# 从环境变量加载私钥
load_dotenv()
private_key = os.getenv("PRIVATE_KEY")# 加密存储方案
from cryptography.fernet import Fernetdef encrypt_private_key(key, private_key):cipher_suite = Fernet(key)return cipher_suite.encrypt(private_key.encode())def decrypt_private_key(key, encrypted_key):cipher_suite = Fernet(key)return cipher_suite.decrypt(encrypted_key).decode()# 生成加密密钥
encryption_key = Fernet.generate_key()# 加密私钥
encrypted_pk = encrypt_private_key(encryption_key, "0xYourPrivateKey")# 安全存储encryption_key和encrypted_pk

6.2 交易安全

# 安全发送交易
def safe_send_transaction(w3, tx, private_key, retries=3):for attempt in range(retries):try:# 更新nonce（每次尝试都需要更新）tx['nonce'] = w3.eth.get_transaction_count(w3.eth.account.from_key(private_key).address)signed_tx = w3.eth.account.sign_transaction(tx, private_key)tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)return w3.eth.wait_for_transaction_receipt(tx_hash)except ValueError as e:if 'nonce too low' in str(e) and attempt < retries - 1:continue  # 重试elif 'insufficient funds' in str(e):raise InsufficientFundsError("账户余额不足")else:raise TransactionError(f"交易失败: {str(e)}")class TransactionError(Exception):passclass InsufficientFundsError(TransactionError):pass

7. 高级技巧与优化

7.1 批量查询

# 批量查询余额
def batch_get_balances(w3, addresses):"""使用批处理RPC请求优化多个余额查询"""from web3 import Web3from web3.middleware import geth_poa_middleware# 创建本地提供者避免阻塞主连接local_w3 = Web3(Web3.HTTPProvider(w3.provider.endpoint_uri))if 'poa' in w3.provider.endpoint_uri:local_w3.middleware_onion.inject(geth_poa_middleware, layer=0)# 使用批处理请求batch = []for address in addresses:batch.append(("eth_getBalance", [address, "latest"]))responses = local_w3.provider.make_batch_request(batch)# 处理响应balances = {}for i, response in enumerate(responses):if response['error'] is not None:balances[addresses[i]] = Noneelse:balances[addresses[i]] = int(response['result'], 16)return balances

7.2 Gas优化策略

# 动态Gas定价
def get_optimal_gas_price(w3):"""获取推荐的Gas价格"""# 方法1：使用历史数据history = w3.eth.fee_history(10, 'latest', [10, 30, 50])base_fee = history['baseFeePerGas'][-1]# 计算优先级费用（小费）的加权平均rewards = [tx[0] for block in history['reward'] for tx in block]if rewards:avg_priority = sum(rewards) / len(rewards)else:avg_priority = w3.to_wei('2', 'gwei')# 方法2：使用预言机（如Etherchain）try:response = requests.get("https://www.etherchain.org/api/gasPriceOracle")data = response.json()fast_gas = w3.to_wei(data['fast'], 'gwei')return min(fast_gas, base_fee * 2 + avg_priority)except:# 回退到基本方法return base_fee * 2 + avg_priority# 在交易中使用
tx = {'to': recipient,'value': amount,'gas': 21000,'maxFeePerGas': get_optimal_gas_price(w3),'maxPriorityFeePerGas': w3.to_wei('2', 'gwei'),'nonce': w3.eth.get_transaction_count(sender),'chainId': w3.eth.chain_id,'type': '0x2'  # EIP-1559类型交易
}

结论：成为区块链开发者

通过本指南，您已掌握Web3.py的核心功能：

1. 基础操作：连接节点、查询余额、发送交易
2. 智能合约：编译、部署和交互
3. 事件处理：监听和响应链上事件
4. DApp开发：完整的前后端集成
5. 安全实践：私钥管理和交易安全
6. 高级技巧：批量查询和Gas优化

Web3.py的强大功能为开发者打开了区块链世界的大门：

• DeFi协议交互：集成Uniswap、Compound等协议
• NFT开发：创建和交易数字藏品
• DAO治理：参与去中心化组织决策
• 跨链应用：桥接不同区块链生态系统

在区块链的世界里，每一行代码都是构建去中心化未来的基石。掌握Web3.py，你就拥有了塑造这个未来的工具。

下一步：

1. 探索Layer2解决方案（Optimism、Arbitrum）
2. 学习智能合约安全审计
3. 研究零知识证明应用
4. 参与开源Web3项目
5. 构建自己的去中心化应用