feat(query): contract creation query support [APE-1296]

.pre-commit-config.yaml

@@ -21,7 +21,7 @@ repos:
     -   id: flake8
 
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v0.991
+    rev: v1.4.1
     hooks:
     -   id: mypy
         additional_dependencies: [
@@ -34,7 +34,7 @@ repos:
         ]
 
 -   repo: https://github.com/executablebooks/mdformat
-    rev: 0.7.14
+    rev: 0.7.16
     hooks:
     -   id: mdformat
         additional_dependencies: [mdformat-gfm, mdformat-frontmatter, mdformat-pyproject]

src/ape/api/providers.py

@@ -388,6 +388,27 @@ def get_transactions_by_account_nonce(  # type: ignore[empty-body]
             Iterator[:class:`~ape.api.transactions.ReceiptAPI`]
         """
 
+    @raises_not_implemented
+    def get_contract_creation_receipts(  # type: ignore[empty-body]
+        self,
+        address: AddressType,
+        start_block: int = 0,
+        stop_block: int = -1,
+        contract_code: Optional[HexBytes] = None,
+    ) -> Iterator[ReceiptAPI]:
+        """
+        Get all receipts where a contract address was created or re-created.
+
+        Args:
+            address (``AddressType``): The address of the account.
+            start_block (int): The block number to start the search with.
+            stop_block (int): The block number to stop the search with.
+            contract_code (Optional[bytes]): The code of the contract at the stop block.
+
+        Returns:
+            Iterator[:class:`~ape.api.transactions.ReceiptAPI`]
+        """
+
     @abstractmethod
     def send_transaction(self, txn: TransactionAPI) -> ReceiptAPI:
         """
@@ -1267,6 +1288,53 @@ def block_ranges(self, start=0, stop=None, page=None):
             stop_block = min(stop, start_block + page - 1)
             yield start_block, stop_block
 
+    def get_contract_creation_receipts(  # type: ignore[empty-body]
+        self,
+        address: AddressType,
+        start_block: int = 0,
+        stop_block: Optional[int] = None,
+        contract_code: Optional[HexBytes] = None,
+    ) -> Iterator[ReceiptAPI]:
+        if stop_block is None:
+            stop_block = self.chain_manager.blocks.height
+
+        if contract_code is None:
+            contract_code = HexBytes(self.get_code(address))
+
+        mid_block = (stop_block - start_block) // 2 + start_block
+        # NOTE: biased towards mid_block == start_block
+
+        if start_block == mid_block:
+            for tx in self.chain_manager.blocks[mid_block].transactions:
+                if (receipt := tx.receipt) and receipt.contract_address == address:
+                    yield receipt
+
+            if mid_block + 1 <= stop_block:
+                yield from self.get_contract_creation_receipts(
+                    address,
+                    start_block=mid_block + 1,
+                    stop_block=stop_block,
+                    contract_code=contract_code,
+                )
+
+        # TODO: Handle when code is nonzero but doesn't match
+        # TODO: Handle when code is empty after it's not (re-init)
+        elif HexBytes(self.get_code(address, block_id=mid_block)) == contract_code:
+            yield from self.get_contract_creation_receipts(
+                address,
+                start_block=start_block,
+                stop_block=mid_block,
+                contract_code=contract_code,
+            )
+
+        elif mid_block + 1 <= stop_block:
+            yield from self.get_contract_creation_receipts(
+                address,
+                start_block=mid_block + 1,
+                stop_block=stop_block,
+                contract_code=contract_code,
+            )
+
     def get_contract_logs(self, log_filter: LogFilter) -> Iterator[ContractLog]:
         height = self.chain_manager.blocks.height
         start_block = log_filter.start_block

src/ape/api/query.py

@@ -13,6 +13,7 @@
     "BlockQuery",
     "BlockTransactionQuery",
     "AccountTransactionQuery",
+    "ContractCreationQuery",
     "ContractEventQuery",
     "ContractMethodQuery",
 ]
@@ -149,6 +150,10 @@ def check_start_nonce_before_stop_nonce(cls, values: Dict) -> Dict:
         return values
 
 
+class ContractCreationQuery(_BaseBlockQuery):
+    contract: AddressType
+
+
 class ContractEventQuery(_BaseBlockQuery):
     """
     A ``QueryType`` that collects members from ``event`` over a range of

src/ape/managers/chain.py

@@ -17,6 +17,7 @@
 from ape.api.query import (
     AccountTransactionQuery,
     BlockQuery,
+    ContractCreationQuery,
     extract_fields,
     validate_and_expand_columns,
 )
@@ -1348,35 +1349,26 @@ def get_creation_receipt(
         Returns:
             :class:`~ape.apt.transactions.ReceiptAPI`
         """
-        if stop_block is None and (stop := self.chain_manager.blocks.head.number):
-            stop_block = stop
-        elif stop_block is None:
-            raise ChainError("Chain missing blocks.")
-
-        mid_block = (stop_block - start_block) // 2 + start_block
-        # NOTE: biased towards mid_block == start_block
-
-        if start_block == mid_block:
-            for tx in self.chain_manager.blocks[mid_block].transactions:
-                if (receipt := tx.receipt) and receipt.contract_address == address:
-                    return receipt
-
-            if mid_block + 1 <= stop_block:
-                return self.get_creation_receipt(
-                    address, start_block=mid_block + 1, stop_block=stop_block
+        if stop_block is None:
+            stop_block = self.chain_manager.blocks.height
+
+        # TODO: Refactor the name of this somehow to be clearer
+        creation_receipts = cast(
+            Iterator[ReceiptAPI],
+            self.query_manager.query(
+                ContractCreationQuery(
+                    columns=["*"],
+                    contract=address,
+                    start_block=start_block,
+                    stop_block=stop_block,
                 )
-            else:
-                raise ChainError(f"Failed to find a contract-creation receipt for '{address}'.")
-
-        elif self.provider.get_code(address, block_id=mid_block):
-            return self.get_creation_receipt(address, start_block=start_block, stop_block=mid_block)
-
-        elif mid_block + 1 <= stop_block:
-            return self.get_creation_receipt(
-                address, start_block=mid_block + 1, stop_block=stop_block
-            )
+            ),
+        )
 
-        else:
+        try:
+            # Get the first contract receipt, which is the first time it appears
+            return next(creation_receipts)
+        except StopIteration:
             raise ChainError(f"Failed to find a contract-creation receipt for '{address}'.")
 
 

src/ape/managers/query.py

@@ -9,6 +9,7 @@
     BaseInterfaceModel,
     BlockQuery,
     BlockTransactionQuery,
+    ContractCreationQuery,
     ContractEventQuery,
 )
 from ape.contracts.base import ContractLog, LogFilter
@@ -38,6 +39,12 @@ def estimate_block_transaction_query(self, query: BlockTransactionQuery) -> int:
         # NOTE: Very loose estimate of 1000ms per block for this query.
         return self.provider.get_block(query.block_id).num_transactions * 100
 
+    @estimate_query.register
+    def estimate_contract_creation_query(self, query: ContractCreationQuery) -> int:
+        # NOTE: Extremely expensive query, involves binary search of all blocks in a chain
+        #       Very loose estimate of 5s per transaction for this query.
+        return 5000
+
     @estimate_query.register
     def estimate_contract_events_query(self, query: ContractEventQuery) -> int:
         # NOTE: Very loose estimate of 100ms per block for this query.
@@ -68,6 +75,14 @@ def perform_block_transaction_query(
     ) -> Iterator[TransactionAPI]:
         return self.provider.get_transactions_by_block(query.block_id)
 
+    @perform_query.register
+    def perform_contract_creation_query(self, query: ContractCreationQuery) -> Iterator[ReceiptAPI]:
+        yield from self.provider.get_contract_creation_receipts(
+            address=query.contract,
+            start_block=query.start_block,
+            stop_block=query.stop_block,
+        )
+
     @perform_query.register
     def perform_contract_events_query(self, query: ContractEventQuery) -> Iterator[ContractLog]:
         addresses = query.contract

src/ape/utils/misc.py

@@ -21,7 +21,7 @@
 from ape.utils.os import expand_environment_variables
 
 EMPTY_BYTES32 = HexBytes("0x0000000000000000000000000000000000000000000000000000000000000000")
-ZERO_ADDRESS = cast(AddressType, "0x0000000000000000000000000000000000000000")
+ZERO_ADDRESS: AddressType = cast(AddressType, "0x0000000000000000000000000000000000000000")
 DEFAULT_TRANSACTION_ACCEPTANCE_TIMEOUT = 120
 DEFAULT_LOCAL_TRANSACTION_ACCEPTANCE_TIMEOUT = 20
 DEFAULT_MAX_RETRIES_TX = 20