Web3.py V6 upgrade

.github/workflows/ruff.yml

@@ -0,0 +1,8 @@
+name: Ruff
+on: [ push, pull_request ]
+jobs:
+  ruff:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/[email protected]
+      - uses: chartboost/ruff-action@v1

.pre-commit-config.yaml

@@ -0,0 +1,10 @@
+repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    # Ruff version.
+    rev: v0.5.4
+    hooks:
+      # Run the linter.
+      - id: ruff
+        args: [--fix]
+      # Run the formatter.
+      - id: ruff-format

.vscode/settings.json

@@ -1,6 +1,10 @@
 {
-  "editor.formatOnSave": true,
   "[python]": {
-    "editor.defaultFormatter": "ms-python.black-formatter"
+    "editor.formatOnSave": true,
+    "editor.codeActionsOnSave": {
+      "source.fixAll": "explicit",
+      "source.organizeImports": "explicit"
+    },
+    "editor.defaultFormatter": "charliermarsh.ruff"
   }
 }

examples/simple.py

@@ -2,26 +2,29 @@
 Minimal viable example of flashbots usage with dynamic fee transactions.
 Sends a bundle of two transactions which transfer some ETH into a random account.
 
-"eth_sendBundle" is a generic method that can be used to send a bundle to any relay.
-For instance, you can use the following relay URLs:
-    titan: 'https://rpc.titanbuilder.xyz/'
-    beaver: 'https://rpc.beaverbuild.org/'
-    builder69: 'https://builder0x69.io/'
-    rsync: 'https://rsync-builder.xyz/'
-    flashbots: 'https://relay.flashbots.net'
-
-You can simply replace the URL in the flashbot method to use a different relay like:
-    flashbot(w3, signer, YOUR_CHOSEN_RELAY_URL)
-
 Environment Variables:
 - ETH_SENDER_KEY: Private key of account which will send the ETH.
-- ETH_SIGNER_KEY: Private key of account which will sign the bundle. 
+- ETH_SIGNER_KEY: Private key of account which will sign the bundle.
     - This account is only used for reputation on flashbots and should be empty.
 - PROVIDER_URL: (Optional) HTTP JSON-RPC Ethereum provider URL. If not set, Flashbots Protect RPC will be used.
+- LOG_LEVEL: (Optional) Set the logging level. Default is 'INFO'. Options: DEBUG, INFO, WARNING, ERROR, CRITICAL.
+
+Usage:
+python examples/simple.py <network> [--log-level LEVEL]
+
+Arguments:
+- network: The network to use (e.g., mainnet, goerli)
+- --log-level: (Optional) Set the logging level. Default is 'INFO'.
+
+Example:
+LOG_LEVEL=DEBUG python examples/simple.py mainnet --log-level DEBUG
 """
 
+import argparse
+import logging
 import os
 import secrets
+from enum import Enum
 from uuid import uuid4
 
 from eth_account.account import Account
@@ -30,163 +33,184 @@
 from web3.exceptions import TransactionNotFound
 from web3.types import TxParams
 
-from flashbots import flashbot
-
-# Define the network to use
-NETWORK = "holesky"  # Options: "sepolia", "holesky", "mainnet"
-
-# Define chain IDs and Flashbots Protect RPC URLs
-NETWORK_CONFIG = {
-    "sepolia": {
-        "chain_id": 11155111,
-        "provider_url": "https://rpc-sepolia.flashbots.net",
-        "relay_url": "https://relay-sepolia.flashbots.net",
-    },
-    "holesky": {
-        "chain_id": 17000,
-        "provider_url": "https://rpc-holesky.flashbots.net",
-        "relay_url": "https://relay-holesky.flashbots.net",
-    },
-    "mainnet": {
-        "chain_id": 1,
-        "provider_url": "https://rpc.flashbots.net",
-        "relay_url": None,  # Mainnet uses default Flashbots relay
-    },
-}
+from flashbots import FlashbotsWeb3, flashbot
+from flashbots.constants import FLASHBOTS_NETWORKS
+from flashbots.types import Network
+
+# Configure logging
+log_level = os.environ.get("LOG_LEVEL", "INFO").upper()
+logging.basicConfig(
+    level=getattr(logging, log_level),
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+class EnumAction(argparse.Action):
+    def __init__(self, **kwargs):
+        enum_type = kwargs.pop("type", None)
+        if enum_type is None:
+            raise ValueError("type must be assigned an Enum when using EnumAction")
+        if not issubclass(enum_type, Enum):
+            raise TypeError("type must be an Enum when using EnumAction")
+        kwargs.setdefault("choices", tuple(e.value for e in enum_type))
+        super(EnumAction, self).__init__(**kwargs)
+        self._enum = enum_type
+
+    def __call__(self, parser, namespace, values, option_string=None):
+        value = self._enum(values)
+        setattr(namespace, self.dest, value)
+
+
+def parse_arguments() -> Network:
+    parser = argparse.ArgumentParser(description="Flashbots simple example")
+    parser.add_argument(
+        "network",
+        type=Network,
+        action=EnumAction,
+        help=f"The network to use ({', '.join(e.value for e in Network)})",
+    )
+    parser.add_argument(
+        "--log-level",
+        type=str,
+        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+        default="INFO",
+        help="Set the logging level",
+    )
+    args = parser.parse_args()
+    return args.network
 
 
 def env(key: str) -> str:
-    return os.environ.get(key)
+    value = os.environ.get(key)
+    if value is None:
+        raise ValueError(f"Environment variable '{key}' is not set")
+    return value
 
 
 def random_account() -> LocalAccount:
     key = "0x" + secrets.token_hex(32)
     return Account.from_key(key)
 
 
-def main() -> None:
-    # account to send the transfer and sign transactions
-    sender: LocalAccount = Account.from_key(env("ETH_SENDER_KEY"))
-    # account to receive the transfer
-    receiverAddress: str = random_account().address
-    # account to sign bundles & establish flashbots reputation
-    # NOTE: this account should not store funds
-    signer: LocalAccount = Account.from_key(env("ETH_SIGNER_KEY"))
-
-    # Use user-provided RPC URL if available, otherwise use Flashbots Protect RPC
-    user_provider_url = env("PROVIDER_URL")
-    if user_provider_url:
-        provider_url = user_provider_url
-        print(f"Using user-provided RPC: {provider_url}")
-    else:
-        provider_url = NETWORK_CONFIG[NETWORK]["provider_url"]
-        print(f"Using Flashbots Protect RPC: {provider_url}")
-
-    w3 = Web3(HTTPProvider(provider_url))
-
-    relay_url = NETWORK_CONFIG[NETWORK]["relay_url"]
-    if relay_url:
-        flashbot(w3, signer, relay_url)
-    else:
-        flashbot(w3, signer)
-
-    print(f"Sender address: {sender.address}")
-    print(f"Receiver address: {receiverAddress}")
-    print(
-        f"Sender account balance: {Web3.from_wei(w3.eth.get_balance(sender.address), 'ether')} ETH"
+def get_account_from_env(key: str) -> LocalAccount:
+    return Account.from_key(env(key))
+
+
+def setup_web3(network: Network) -> FlashbotsWeb3:
+    provider_url = os.environ.get(
+        "PROVIDER_URL", FLASHBOTS_NETWORKS[network]["provider_url"]
     )
-    print(
-        f"Receiver account balance: {Web3.from_wei(w3.eth.get_balance(receiverAddress), 'ether')} ETH"
+    logger.info(f"Using RPC: {provider_url}")
+    relay_url = FLASHBOTS_NETWORKS[network]["relay_url"]
+    w3 = flashbot(
+        Web3(HTTPProvider(provider_url)),
+        get_account_from_env("ETH_SIGNER_KEY"),
+        relay_url,
     )
+    return w3
 
-    # bundle two EIP-1559 (type 2) transactions, pre-sign one of them
-    # NOTE: chainId is necessary for all EIP-1559 txns
-    # NOTE: nonce is required for signed txns
 
-    nonce = w3.eth.get_transaction_count(sender.address)
-    tx1: TxParams = {
-        "to": receiverAddress,
-        "value": Web3.to_wei(0.001, "ether"),
+def log_account_balances(w3: Web3, sender: str, receiver: str) -> None:
+    logger.info(
+        f"Sender account balance: {Web3.from_wei(w3.eth.get_balance(Web3.to_checksum_address(sender)), 'ether')} ETH"
+    )
+    logger.info(
+        f"Receiver account balance: {Web3.from_wei(w3.eth.get_balance(Web3.to_checksum_address(receiver)), 'ether')} ETH"
+    )
+
+
+def create_transaction(
+    w3: Web3, sender: str, receiver: str, nonce: int, network: Network
+) -> TxParams:
+    # Get the latest gas price information
+    latest = w3.eth.get_block("latest")
+    base_fee = latest["baseFeePerGas"]
+
+    # Set max priority fee (tip) to 2 Gwei
+    max_priority_fee = Web3.to_wei(2, "gwei")
+
+    # Set max fee to be base fee + priority fee
+    max_fee = base_fee + max_priority_fee
+
+    return {
+        "from": sender,
+        "to": receiver,
         "gas": 21000,
-        "maxFeePerGas": Web3.to_wei(200, "gwei"),
-        "maxPriorityFeePerGas": Web3.to_wei(50, "gwei"),
+        "value": Web3.to_wei(0.001, "ether"),
         "nonce": nonce,
-        "chainId": NETWORK_CONFIG[NETWORK]["chain_id"],
-        "type": 2,
+        "maxFeePerGas": max_fee,
+        "maxPriorityFeePerGas": max_priority_fee,
+        "chainId": FLASHBOTS_NETWORKS[network]["chain_id"],
     }
-    tx1_signed = sender.sign_transaction(tx1)
 
-    tx2: TxParams = {
-        "to": receiverAddress,
-        "value": Web3.to_wei(0.001, "ether"),
-        "gas": 21000,
-        "maxFeePerGas": Web3.to_wei(200, "gwei"),
-        "maxPriorityFeePerGas": Web3.to_wei(50, "gwei"),
-        "nonce": nonce + 1,
-        "chainId": NETWORK_CONFIG[NETWORK]["chain_id"],
-        "type": 2,
-    }
 
+def main() -> None:
+    network = parse_arguments()
+    sender = get_account_from_env("ETH_SENDER_KEY")
+    receiver = Account.create().address
+    w3 = setup_web3(network)
+
+    logger.info(f"Sender address: {sender.address}")
+    logger.info(f"Receiver address: {receiver}")
+    log_account_balances(w3, sender.address, receiver)
+
+    nonce = w3.eth.get_transaction_count(sender.address)
+    tx1 = create_transaction(w3, sender.address, receiver, nonce, network)
+    tx2 = create_transaction(w3, sender.address, receiver, nonce + 1, network)
+
+    tx1_signed = w3.eth.account.sign_transaction(tx1, private_key=sender.key)
     bundle = [
         {"signed_transaction": tx1_signed.rawTransaction},
-        {"signer": sender, "transaction": tx2},
+        {"transaction": tx2, "signer": sender},
     ]
 
     # keep trying to send bundle until it gets mined
     while True:
         block = w3.eth.block_number
 
         # Simulation is only supported on mainnet
-        if NETWORK == "mainnet":
-            print(f"Simulating on block {block}")
+        if network == "mainnet":
             # Simulate bundle on current block.
             # If your RPC provider is not fast enough, you may get "block extrapolation negative"
             # error message triggered by "extrapolate_timestamp" function in "flashbots.py".
             try:
                 w3.flashbots.simulate(bundle, block)
-                print("Simulation successful.")
             except Exception as e:
-                print("Simulation error", e)
+                logger.error(f"Simulation error: {e}")
                 return
 
         # send bundle targeting next block
-        print(f"Sending bundle targeting block {block+1}")
         replacement_uuid = str(uuid4())
-        print(f"replacementUuid {replacement_uuid}")
+        logger.info(f"replacementUuid {replacement_uuid}")
         send_result = w3.flashbots.send_bundle(
             bundle,
             target_block_number=block + 1,
             opts={"replacementUuid": replacement_uuid},
         )
-        print("bundleHash", w3.to_hex(send_result.bundle_hash()))
+        logger.info(f"bundleHash {w3.to_hex(send_result.bundle_hash())}")
 
         stats_v1 = w3.flashbots.get_bundle_stats(
             w3.to_hex(send_result.bundle_hash()), block
         )
-        print("bundleStats v1", stats_v1)
+        logger.info(f"bundleStats v1 {stats_v1}")
 
         stats_v2 = w3.flashbots.get_bundle_stats_v2(
             w3.to_hex(send_result.bundle_hash()), block
         )
-        print("bundleStats v2", stats_v2)
+        logger.info(f"bundleStats v2 {stats_v2}")
 
         send_result.wait()
         try:
             receipts = send_result.receipts()
-            print(f"\nBundle was mined in block {receipts[0].blockNumber}\a")
+            logger.info(f"Bundle was mined in block {receipts[0].blockNumber}")
             break
         except TransactionNotFound:
-            print(f"Bundle not found in block {block+1}")
-            # essentially a no-op but it shows that the function works
+            logger.info(f"Bundle not found in block {block + 1}")
             cancel_res = w3.flashbots.cancel_bundles(replacement_uuid)
-            print(f"canceled {cancel_res}")
+            logger.info(f"Canceled {cancel_res}")
 
-    print(
-        f"Sender account balance: {Web3.from_wei(w3.eth.get_balance(sender.address), 'ether')} ETH"
-    )
-    print(
-        f"Receiver account balance: {Web3.from_wei(w3.eth.get_balance(receiverAddress), 'ether')} ETH"
-    )
+    log_account_balances(w3, sender.address, receiver)
 
 
 if __name__ == "__main__":