Web3.py¶
Introduction¶
Interacting with blockchains typically requires an interface between your application and the network. Web3.py offers this interface through a collection of libraries, facilitating seamless interaction with the nodes using HTTP or WebSocket protocols.
This guide illustrates how to utilize Web3.py for interactions with Polkadot Hub.
Set Up the Project¶
-
To start working with Web3.py, begin by initializing your project:
-
Create and activate a virtual environment for your project:
-
Next, install the Web3.py library:
Set Up the Web3 Provider¶
The provider configuration is the foundation of any Web3.py application. It serves as a bridge between your application and the blockchain, allowing you to query blockchain data and interact with smart contracts.
To interact with Polkadot Hub, you must set up a Web3.py provider. This provider connects to a blockchain node, allowing you to query blockchain data and interact with smart contracts. The following code sets up the provider configuration:
Note
Replace INSERT_RPC_URL with the appropriate value. For instance, to connect to Polkadot Hub TestNet, use the following parameter:
With the Web3 provider set up, start querying the blockchain. For instance, you can use the following code snippet to fetch the latest block number of the chain.
Fetch last block example
from web3 import Web3
def main():
try:
PROVIDER_RPC = "https://services.polkadothub-rpc.com/testnet"
web3 = Web3(Web3.HTTPProvider(PROVIDER_RPC))
latest_block = web3.eth.block_number
print("Last block: " + str(latest_block))
except Exception as error:
print("Error connecting to Polkadot Hub TestNet: " + str(error))
if __name__ == "__main__":
main()
Compile Contracts¶
Polkadot Hub exposes an Ethereum JSON-RPC endpoint, so you can compile Solidity contracts to familiar EVM bytecode with the py-solc-x compiler. The resulting artifacts work with any EVM-compatible toolchain and can be deployed through Web3.py.
First, install the py-solc-x package:
Sample Storage Smart Contract¶
This example demonstrates compiling a Storage.sol Solidity contract for deployment to Polkadot Hub. The contract's functionality stores a number and permits users to update it with a new value.
//SPDX-License-Identifier: MIT
// Solidity files have to start with this pragma.
// It will be used by the Solidity compiler to validate its version.
pragma solidity ^0.8.9;
contract Storage {
// Public state variable to store a number
uint256 public storedNumber;
/**
* Updates the stored number.
*
* The `public` modifier allows anyone to call this function.
*
* @param _newNumber - The new value to store.
*/
function setNumber(uint256 _newNumber) public {
storedNumber = _newNumber;
}
}
Compile the Smart Contract¶
To compile this contract, create a Python script named compile.py:
import json
import solcx
from pathlib import Path
SOLC_VERSION = '0.8.9'
try:
solcx.install_solc(SOLC_VERSION)
except Exception as e:
print(f"Solc version {SOLC_VERSION} already installed or error: {e}")
solcx.set_solc_version(SOLC_VERSION)
contract_path = Path('Storage.sol')
with open(contract_path, 'r') as file:
contract_source = file.read()
compiled_sol = solcx.compile_source(
contract_source,
output_values=['abi', 'bin'],
solc_version=SOLC_VERSION
)
contract_id, contract_interface = compiled_sol.popitem()
bytecode = contract_interface['bin']
abi = contract_interface['abi']
Path('abis').mkdir(exist_ok=True)
Path('artifacts').mkdir(exist_ok=True)
with open('abis/Storage.json', 'w') as abi_file:
json.dump(abi, abi_file, indent=2)
with open('artifacts/Storage.bin', 'w') as bin_file:
bin_file.write(bytecode)
print("✅ Contract compiled successfully!")
print(f"📄 ABI saved to: abis/Storage.json")
print(f"📦 Bytecode saved to: artifacts/Storage.bin")
Note
The script above is tailored to the Storage.sol contract. It can be adjusted for other contracts by changing the file name or modifying the ABI and bytecode paths.
The ABI (Application Binary Interface) is a JSON representation of your contract's functions, events, and their parameters. It serves as the interface between your Python code and the deployed smart contract, allowing your application to know how to format function calls and interpret returned data.
Execute the script by running:
After executing the script, the Solidity contract is compiled into standard EVM bytecode. The ABI and bytecode are saved into files with .json and .bin extensions, respectively:
- ABI file (
abis/Storage.json): Provides a JSON interface describing the contract's functions and how to interact with it. - Bytecode file (
artifacts/Storage.bin): Contains the low-level machine code executable on EVM that represents the compiled smart contract ready for blockchain deployment.
You can now proceed with deploying the contract to Polkadot Hub, as outlined in the next section.
Contract Deployment¶
To deploy your compiled contract to Polkadot Hub using Web3.py, you'll need an account with a private key to sign the deployment transaction. The deployment process is exactly the same as for any Ethereum-compatible chain, involving creating a contract instance, estimating gas, and sending a deployment transaction. Here's how to deploy the contract. Replace INSERT_RPC_URL and INSERT_PRIVATE_KEY with the appropriate values:
from web3 import Web3
import json
import time
from pathlib import Path
ARTIFACTS_DIR = Path(__file__).parent
ABI_DIR = ARTIFACTS_DIR / "abis"
BYTECODE_DIR = ARTIFACTS_DIR / "artifacts"
def get_abi(contract_name):
try:
with open(ABI_DIR / f"{contract_name}.json", 'r') as file:
return json.load(file)
except Exception as error:
print(f"❌ Could not find ABI for contract {contract_name}: {error}")
raise error
def get_bytecode(contract_name):
try:
with open(BYTECODE_DIR / f"{contract_name}.bin", 'r') as file:
bytecode = file.read().strip()
return bytecode if bytecode.startswith('0x') else f"0x{bytecode}"
except Exception as error:
print(f"❌ Could not find bytecode for contract {contract_name}: {error}")
raise error
def deploy_with_retry(config, max_retries=3):
"""Deploy with retry logic for RPC errors"""
for attempt in range(max_retries):
try:
return deploy(config)
except Exception as error:
error_str = str(error)
if "500" in error_str or "Internal Server Error" in error_str or "Connection" in error_str:
if attempt < max_retries - 1:
wait_time = (attempt + 1) * 3
print(f"RPC error, retrying in {wait_time} seconds... (attempt {attempt + 1}/{max_retries})")
time.sleep(wait_time)
continue
raise error
def deploy(config):
try:
# Initialize Web3 with RPC URL and longer timeout
web3 = Web3(Web3.HTTPProvider(
config["rpc_url"],
request_kwargs={'timeout': 120}
))
# Prepare account
formatted_private_key = config["private_key"] if config["private_key"].startswith('0x') else f"0x{config['private_key']}"
account = web3.eth.account.from_key(formatted_private_key)
print(f"Deploying from address: {account.address}")
# Load ABI and bytecode
abi = get_abi('Storage')
bytecode = get_bytecode('Storage')
print(f"Bytecode length: {len(bytecode)}")
# Create contract instance
contract = web3.eth.contract(abi=abi, bytecode=bytecode)
# Get current nonce (this will test the connection)
print("Getting nonce...")
nonce = web3.eth.get_transaction_count(account.address)
print(f"Nonce: {nonce}")
# Estimate gas
print("Estimating gas...")
gas_estimate = web3.eth.estimate_gas({
'from': account.address,
'data': bytecode
})
print(f"Estimated gas: {gas_estimate}")
# Get gas price
print("Getting gas price...")
gas_price = web3.eth.gas_price
print(f"Gas price: {web3.from_wei(gas_price, 'gwei')} gwei")
# Build deployment transaction
print("Building transaction...")
construct_txn = contract.constructor().build_transaction({
'from': account.address,
'nonce': nonce,
'gas': gas_estimate,
'gasPrice': gas_price,
})
# Sign transaction
print("Signing transaction...")
signed_txn = web3.eth.account.sign_transaction(construct_txn, private_key=formatted_private_key)
# Send transaction
print("Sending transaction...")
tx_hash = web3.eth.send_raw_transaction(signed_txn.raw_transaction)
print(f"Transaction hash: {tx_hash.hex()}")
# Wait for transaction receipt
print("Waiting for transaction receipt...")
tx_receipt = web3.eth.wait_for_transaction_receipt(tx_hash, timeout=300)
contract_address = tx_receipt.contractAddress
# Log results
print(f"✅ Contract deployed at: {contract_address}")
print(f"Gas used: {tx_receipt.gasUsed}")
print(f"Block number: {tx_receipt.blockNumber}")
return web3.eth.contract(address=contract_address, abi=abi)
except Exception as error:
print(f'❌ Deployment failed: {error}')
raise error
if __name__ == "__main__":
deployment_config = {
"rpc_url": "https://services.polkadothub-rpc.com/testnet",
"private_key": "INSERT_PRIVATE_KEY"
}
deploy_with_retry(deployment_config)
Warning
Never commit or share your private key. Exposed keys can lead to immediate theft of all associated funds.
To run the script, execute the following command:
After running this script, your contract will be deployed to Polkadot Hub, and its address will be printed in your terminal. You can use this address for future contract interactions.
Interact with the Contract¶
After deployment, interact with your contract using Web3.py methods. The example below demonstrates how to set and retrieve a number.
from web3 import Web3
import json
def get_abi(contract_name):
try:
with open(f"{contract_name}.json", "r") as file:
return json.load(file)
except Exception as error:
print(f"❌ Could not find ABI for contract {contract_name}: {error}")
raise error
async def update_storage(config):
try:
# Initialize Web3 with RPC URL
web3 = Web3(Web3.HTTPProvider(config["rpc_url"]))
# Prepare account
account = web3.eth.account.from_key(config["private_key"])
# Load ABI
abi = get_abi("Storage")
# Create contract instance
contract = web3.eth.contract(address=config["contract_address"], abi=abi)
# Get initial value
initial_value = contract.functions.storedNumber().call()
print("Current stored value:", initial_value)
# Get current nonce
nonce = web3.eth.get_transaction_count(account.address)
# Prepare transaction
transaction = contract.functions.setNumber(1).build_transaction(
{"from": account.address, "nonce": nonce}
)
# Sign transaction
signed_txn = web3.eth.account.sign_transaction(
transaction, private_key=config["private_key"]
)
# Send transaction
tx_hash = web3.eth.send_raw_transaction(signed_txn.raw_transaction)
print(f"Transaction hash: {tx_hash.hex()}")
# Wait for receipt
receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
# Get updated value
new_value = contract.functions.storedNumber().call()
print("New stored value:", new_value)
return receipt
except Exception as error:
print("Update failed:", error)
raise error
if __name__ == "__main__":
# Example usage
import asyncio
config = {
"rpc_url": "INSERT_RPC_URL",
"private_key": "INSERT_PRIVATE_KEY",
"contract_address": "INSERT_CONTRACT_ADDRESS",
}
asyncio.run(update_storage(config))
Be sure to replace the INSERT_RPC_URL, INSERT_PRIVATE_KEY, and INSERT_CONTRACT_ADDRESS placeholders with your specific values.
To interact with the contract, run:
Where to Go Next¶
-
External Web3.py Docs
Explore the Web3.py documentation to learn how to use additional features, such as wallet management, signing messages, subscribing to events, and more.
| Created: January 14, 2026