Modify _get_receipt to accept a tx hash so that it can be reused.

Allow FinalizedTx to have a txhash property - makes it easier for using/logging txhash in a variety of areas including methods that take a Union of PendingTx and FinalizedTx.

atxm/machine.py

@@ -383,7 +383,7 @@ def __get_receipt(self, pending_tx: PendingTx) -> Optional[TxReceipt]:
             )
             return
 
-        receipt = _get_receipt(w3=self.w3, data=txdata)
+        receipt = _get_receipt(w3=self.w3, txhash=txdata['hash'])
         if not receipt:
             return
 
@@ -405,32 +405,28 @@ def __get_receipt(self, pending_tx: PendingTx) -> Optional[TxReceipt]:
 
     def __get_confirmations(self, tx: Union[PendingTx, FinalizedTx]) -> int:
         current_block = self.w3.eth.block_number
-        txhash = (
-            tx.txhash if isinstance(tx, PendingTx) else tx.receipt["transactionHash"]
-        )
-        try:
-            tx_receipt = self.w3.eth.get_transaction_receipt(txhash)
-            tx_block = tx_receipt["blockNumber"]
-            confirmations = current_block - tx_block
-            return confirmations
-        except TransactionNotFound:
-            self.log.info(f"Transaction {txhash.hex()} is pending or unconfirmed")
+        tx_receipt = _get_receipt(w3=self.w3, txhash=tx.txhash)
+        if not tx_receipt:
+            self.log.info(f"Transaction {tx.txhash.hex()} is pending or unconfirmed")
             return 0
 
+        tx_block = tx_receipt["blockNumber"]
+        confirmations = current_block - tx_block
+        return confirmations
+
     def __monitor_finalized(self) -> None:
         """Follow-up on finalized transactions for a little while."""
         if not self._state.finalized:
             return
         for tx in self._state.finalized.copy():
             confirmations = self.__get_confirmations(tx=tx)
-            txhash = tx.receipt["transactionHash"]
             if confirmations >= self._TRACKING_CONFIRMATIONS:
                 if tx in self._state.finalized:
                     self._state.finalized.remove(tx)
                     self.log.info(
-                        f"[clear] stopped tracking {txhash.hex()} after {confirmations} confirmations"
+                        f"[clear] stopped tracking {tx.txhash.hex()} after {confirmations} confirmations"
                     )
                 continue
             self.log.info(
-                f"[monitor] transaction {txhash.hex()} has {confirmations} confirmations"
+                f"[monitor] transaction {tx.txhash.hex()} has {confirmations} confirmations"
             )

atxm/tx.py

@@ -115,6 +115,10 @@ def from_dict(cls, data: Dict):
         receipt = _deserialize_tx_receipt(data["receipt"])
         return cls(id=int(data["id"]), receipt=receipt)
 
+    @property
+    def txhash(self) -> TxHash:
+        return self.receipt["transactionHash"]
+
 
 @dataclass
 class FaultyTx(AsyncTx):

atxm/utils.py

@@ -12,7 +12,7 @@
     InsufficientFunds,
 )
 from atxm.logging import log
-from atxm.tx import AsyncTx, FutureTx
+from atxm.tx import AsyncTx, FutureTx, TxHash
 
 
 @memoize
@@ -36,9 +36,9 @@ def _log_gas_weather(base_fee: Wei, tip: Wei) -> None:
     log.info(f"Gas conditions: base {base_fee_gwei} gwei | tip {tip_gwei} gwei")
 
 
-def _get_receipt(w3: Web3, data: Union[TxData, PendingTxData]) -> Optional[TxReceipt]:
+def _get_receipt(w3: Web3, txhash: TxHash) -> Optional[TxReceipt]:
     try:
-        receipt = w3.eth.get_transaction_receipt(data["hash"])
+        receipt = w3.eth.get_transaction_receipt(txhash)
     except TransactionNotFound:
         return
     return receipt