Added CLI for moonstream.abi_decoder

backend/moonstream/abi_decoder.py

@@ -1,8 +1,12 @@
-from typing import Generic, List, Optional, TypeVar, Union, Type, cast
-import pyevmasm
+import argparse
 import binascii
+import json
+import sys
+from typing import List, Optional, Union, Type, cast
 
-from moonstreamdb.db import yield_db_session_ctx
+import pyevmasm
+
+from moonstreamdb.db import yield_db_session
 from moonstreamdb.models import ESDEventSignature, ESDFunctionSignature
 from sqlalchemy.orm.session import Session
 from sqlalchemy.sql.expression import text
@@ -41,11 +45,16 @@ def decode_signatures(
     return decoded_signatures
 
 
-def decode_abi(source: str) -> ContractABI:
+def decode_abi(source: str, session: Optional[Session] = None) -> ContractABI:
     disassembled = pyevmasm.disassemble_all(binascii.unhexlify(source))
     function_hex_signatures = []
     event_hex_signatures = []
 
+    should_close_session = False
+    if session is None:
+        should_close_session = True
+        session = next(yield_db_session())
+
     for instruction in disassembled:
         if instruction.name == "PUSH4":
             hex_signature = "0x{:x}".format(instruction.operand)
@@ -56,16 +65,42 @@ def decode_abi(source: str) -> ContractABI:
             if hex_signature not in event_hex_signatures:
                 event_hex_signatures.append(hex_signature)
 
-    with yield_db_session_ctx() as session:
+    try:
         function_signatures = decode_signatures(
             session, function_hex_signatures, EVMFunctionSignature, ESDFunctionSignature
         )
         event_signatures = decode_signatures(
             session, event_hex_signatures, EVMEventSignature, ESDEventSignature
         )
+    finally:
+        if should_close_session:
+            session.close()
 
-        abi = ContractABI(
-            functions=cast(EVMFunctionSignature, function_signatures),
-            events=cast(EVMEventSignature, event_signatures),
-        )
-        return abi
+    abi = ContractABI(
+        functions=cast(EVMFunctionSignature, function_signatures),
+        events=cast(EVMEventSignature, event_signatures),
+    )
+    return abi
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Decode Ethereum smart contract ABIs")
+    parser.add_argument(
+        "-i",
+        "--infile",
+        type=argparse.FileType("r"),
+        default=sys.stdin,
+        help="File containing the ABI to decode",
+    )
+    args = parser.parse_args()
+
+    source: Optional[str] = None
+    with args.infile as ifp:
+        source = ifp.read().strip()
+
+    abi = decode_abi(source)
+    print(abi.json())
+
+
+if __name__ == "__main__":
+    main()