feat: type 3 tx initial support (#1928)

src/ape_ethereum/ecosystem.py

@@ -62,6 +62,8 @@
     BaseTransaction,
     DynamicFeeTransaction,
     Receipt,
+    SharedBlobReceipt,
+    SharedBlobTransaction,
     StaticFeeTransaction,
     TransactionStatusEnum,
     TransactionType,
@@ -480,7 +482,7 @@ def str_to_slot(text):
 
     def decode_receipt(self, data: Dict) -> ReceiptAPI:
         status = data.get("status")
-        if status:
+        if status is not None:
             status = self.conversion_manager.convert(status, int)
             status = TransactionStatusEnum(status)
 
@@ -505,7 +507,7 @@ def decode_receipt(self, data: Dict) -> ReceiptAPI:
         if block_number is None:
             raise ValueError("Missing block number.")
 
-        receipt = Receipt(
+        receipt_kwargs = dict(
             block_number=block_number,
             contract_address=data.get("contract_address") or data.get("contractAddress"),
             gas_limit=data.get("gas", data.get("gas_limit", data.get("gasLimit"))) or 0,
@@ -516,7 +518,19 @@ def decode_receipt(self, data: Dict) -> ReceiptAPI:
             txn_hash=txn_hash,
             transaction=self.create_transaction(**data),
         )
-        return receipt
+
+        receipt_cls: Type[Receipt]
+        if any(
+            x in data
+            for x in ("blobGasPrice", "blobGasUsed", "blobVersionedHashes", "maxFeePerBlobGas")
+        ):
+            receipt_cls = SharedBlobReceipt
+            receipt_kwargs["blob_gas_price"] = data.get("blob_gas_price", data.get("blobGasPrice"))
+            receipt_kwargs["blob_gas_used"] = data.get("blob_gas_used", data.get("blobGasUsed"))
+        else:
+            receipt_cls = Receipt
+
+        return receipt_cls.model_validate(receipt_kwargs)
 
     def decode_block(self, data: Dict) -> BlockAPI:
         data["hash"] = HexBytes(data["hash"]) if data.get("hash") else None
@@ -768,6 +782,8 @@ def create_transaction(self, **kwargs) -> TransactionAPI:
             tx_data,
             ("txType", "tx_type", "txnType", "txn_type", "transactionType", "transaction_type"),
         )
+        tx_data = _correct_key("maxFeePerBlobGas", tx_data, ("max_fee_per_blob_gas",))
+        tx_data = _correct_key("blobVersionedHashes", tx_data, ("blob_versioned_hashes",))
 
         # Handle unique value specifications, such as "1 ether".
         if "value" in tx_data and not isinstance(tx_data["value"], int):
@@ -781,8 +797,9 @@ def create_transaction(self, **kwargs) -> TransactionAPI:
         # Deduce the transaction type.
         transaction_types: Dict[TransactionType, Type[TransactionAPI]] = {
             TransactionType.STATIC: StaticFeeTransaction,
-            TransactionType.DYNAMIC: DynamicFeeTransaction,
             TransactionType.ACCESS_LIST: AccessListTransaction,
+            TransactionType.DYNAMIC: DynamicFeeTransaction,
+            TransactionType.SHARED_BLOB: SharedBlobTransaction,
         }
         if "type" in tx_data:
             # May be None in data.
@@ -803,6 +820,8 @@ def create_transaction(self, **kwargs) -> TransactionAPI:
             version = TransactionType.DYNAMIC
         elif "access_list" in tx_data or "accessList" in tx_data:
             version = TransactionType.ACCESS_LIST
+        elif "maxFeePerBlobGas" in tx_data or "blobVersionedHashes" in tx_data:
+            version = TransactionType.SHARED_BLOB
         else:
             version = self.default_transaction_type
 

src/ape_ethereum/provider.py

@@ -830,7 +830,8 @@ def prepare_transaction(self, txn: TransactionAPI) -> TransactionAPI:
             and txn.gas_price is None
         ):
             txn.gas_price = self.gas_price
-        elif txn_type is TransactionType.DYNAMIC:
+
+        elif txn_type in (TransactionType.DYNAMIC, TransactionType.SHARED_BLOB):
             if txn.max_priority_fee is None:
                 txn.max_priority_fee = self.priority_fee
 

src/ape_ethereum/transactions.py

@@ -47,6 +47,7 @@ class TransactionType(Enum):
     STATIC = 0
     ACCESS_LIST = 1  # EIP-2930
     DYNAMIC = 2  # EIP-1559
+    SHARED_BLOB = 3  # EIP-4844
 
 
 class AccessList(BaseModel):
@@ -128,6 +129,22 @@ def calculate_read_only_max_fee(cls, values) -> Dict:
         return values
 
 
+class AccessListTransaction(StaticFeeTransaction):
+    """
+    `EIP-2930 <https://eips.ethereum.org/EIPS/eip-2930>`__
+    transactions are similar to legacy transaction with an added access list functionality.
+    """
+
+    gas_price: Optional[int] = Field(None, alias="gasPrice")
+    type: int = Field(TransactionType.ACCESS_LIST.value)
+    access_list: List[AccessList] = Field(default_factory=list, alias="accessList")
+
+    @field_validator("type")
+    @classmethod
+    def check_type(cls, value):
+        return value.value if isinstance(value, TransactionType) else value
+
+
 class DynamicFeeTransaction(BaseTransaction):
     """
     Transactions that are post-EIP-1559 and use the ``maxFeePerGas``
@@ -145,20 +162,23 @@ def check_type(cls, value):
         return value.value if isinstance(value, TransactionType) else value
 
 
-class AccessListTransaction(StaticFeeTransaction):
+class SharedBlobTransaction(DynamicFeeTransaction):
     """
-    `EIP-2930 <https://eips.ethereum.org/EIPS/eip-2930>`__
-    transactions are similar to legacy transaction with an added access list functionality.
+    `EIP-4844 <https://eips.ethereum.org/EIPS/eip-4844>`__ transactions.
     """
 
-    gas_price: Optional[int] = Field(None, alias="gasPrice")
-    type: int = Field(TransactionType.ACCESS_LIST.value)
-    access_list: List[AccessList] = Field(default_factory=list, alias="accessList")
+    max_fee_per_blob_gas: int = Field(0, alias="maxFeePerBlobGas")
+    blob_versioned_hashes: List[HexBytes] = Field([], alias="blobVersionedHashes")
 
-    @field_validator("type")
+    """
+    Overridden because EIP-4844 states it cannot be nil.
+    """
+    receiver: AddressType = Field(ZERO_ADDRESS, alias="to")
+
+    @field_validator("max_fee_per_blob_gas", mode="before")
     @classmethod
-    def check_type(cls, value):
-        return value.value if isinstance(value, TransactionType) else value
+    def hex_to_int(cls, value):
+        return value if isinstance(value, int) else int(HexBytes(value).hex(), 16)
 
 
 class Receipt(ReceiptAPI):
@@ -404,3 +424,13 @@ def _decode_ds_note(self, log: Dict) -> Optional[ContractLog]:
             transaction_hash=log["transactionHash"],
             transaction_index=log["transactionIndex"],
         )
+
+
+class SharedBlobReceipt(Receipt):
+    blob_gas_used: int
+    blob_gas_price: int
+
+    @field_validator("blob_gas_used", "blob_gas_price", mode="before")
+    @classmethod
+    def hex_to_int(cls, value):
+        return value if isinstance(value, int) else int(HexBytes(value).hex(), 16)