feat(draft): add deployment logs to network mode

boa/deployments.py

@@ -0,0 +1,81 @@
+from dataclasses import dataclass, asdict
+from boa.util.open_ctx import Open
+from pathlib import Path
+from typing import Optional,Any
+import json
+import sqlite3
+from boa.util.abi import Address
+
+@dataclass(frozen=True)
+class Deployment:
+    contract_address: Address
+    name: str
+    rpc: str
+    from_: Address
+    tx_hash: str
+    broadcast_ts: float
+    tx_dict: dict  # raw tx fields
+    receipt_dict: dict  # raw receipt fields
+    source_code: Optional[Any]  # optional source code or bundle
+
+    def sql_values(self):
+        ret = asdict(self)
+        ret["contract_address"] = str(ret["contract_address"])
+        ret["from_"] = str(ret["from_"])
+        ret["tx_hash"] = ret["tx_hash"]
+        ret["tx_dict"] = json.dumps(ret["tx_dict"])
+        ret["receipt_dict"] = json.dumps(ret["receipt_dict"])
+        if ret["source_code"] is not None:
+            ret["source_code"] = json.dumps(ret["source_code"])
+        return ret
+
+
+_CREATE_CMD = """
+CREATE TABLE IF NOT EXISTS
+    deployments(
+        contract_address text,
+        name text,
+        rpc text,
+        tx_hash text,
+        from_ text,
+        tx_dict text,
+        broadcast_ts real,
+        receipt_dict text,
+        source_code text
+    );
+"""
+
+class DeploymentsDB:
+    def __init__(self, path="./.boa/deployments.db"):
+        path = Path(path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+
+        # once 3.12 is min version, use autocommit=True
+        self.db = sqlite3.connect(path)
+
+        self.db.execute(_CREATE_CMD)
+
+    def __del__(self):
+        self.db.close()
+
+    def insert_deployment(self, deployment: Deployment):
+        values = deployment.sql_values()
+
+        values_placeholder = ",".join(["?"] * len(values))
+
+        insert_cmd = f"INSERT INTO deployments VALUES({values_placeholder});"
+
+        self.db.execute(insert_cmd, tuple(values.values()))
+        self.db.commit()
+
+_db: Optional[DeploymentsDB] = None
+
+def set_deployments_db(db: Optional[DeploymentsDB]):
+    def set_(db):
+        global _db
+        _db = db
+    return Open(get_deployments_db, set_, db)
+
+def get_deployments_db():
+    global _db
+    return _db

boa/network.py

@@ -1,5 +1,6 @@
 # an Environment which interacts with a real (prod or test) chain
 import contextlib
+import time
 import warnings
 from dataclasses import dataclass
 from functools import cached_property
@@ -8,6 +9,7 @@
 from eth_account import Account
 from requests.exceptions import HTTPError
 
+from boa.deployments import Deployment, get_deployments_db
 from boa.environment import Env, _AddressType
 from boa.rpc import (
     RPC,
@@ -300,7 +302,7 @@ def execute_code(
 
         if is_modifying:
             try:
-                receipt, trace = self._send_txn(
+                txdata, receipt, trace = self._send_txn(
                     from_=sender, to=to_address, value=value, gas=gas, data=hexdata
                 )
             except _EstimateGasFailed:
@@ -375,7 +377,9 @@ def deploy(
         bytecode = to_hex(bytecode)
         sender = self._check_sender(self._get_sender(sender))
 
-        receipt, trace = self._send_txn(
+        broadcast_ts = time.time()
+
+        txdata, receipt, trace = self._send_txn(
             from_=sender, value=value, gas=gas, data=bytecode
         )
 
@@ -397,6 +401,21 @@ def deploy(
         # TODO get contract info in here
         print(f"contract deployed at {create_address}")
 
+        if (deployments_db := get_deployments_db()) is not None:
+            contract_name = getattr(contract, "contract_name", None)
+            deployment_data = Deployment(
+                create_address,
+                contract_name,
+                self._rpc.identifier,
+                sender,
+                receipt["transactionHash"],
+                broadcast_ts,
+                txdata,
+                receipt,
+                None,
+            )
+            deployments_db.insert_deployment(deployment_data)
+
         return create_address, computation
 
     @cached_property
@@ -538,7 +557,7 @@ def _send_txn(self, from_, to=None, gas=None, value=None, data=None):
         self._reset_fork(block_identifier=receipt["blockNumber"])
 
         t_obj = TraceObject(trace) if trace is not None else None
-        return receipt, t_obj
+        return tx_data, receipt, t_obj
 
     def get_chain_id(self) -> int:
         """Get the current chain ID of the network as an integer."""