Merge branch 'main' into extend-subgraph/ecosystems

.github/dependabot.yaml

@@ -0,0 +1,9 @@
+# ref: https://docs.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
+
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    # Workflow files stored in the default location of `.github/workflows`. (You don't need to specify `/.github/workflows` for `directory`. You can use `directory: "/"`.)
+    directory: "/"
+    schedule:
+      interval: "daily"

.github/workflows/python_package.yaml

@@ -0,0 +1,29 @@
+name: Python package
+
+on: [pull_request]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.10"]
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: "pip"
+      # You can test your matrix by printing the current Python version
+      - name: Display Python version
+        run: python -c "import sys; print(sys.version)"
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r bal_tools/requirements.txt
+      - name: Test with pytest
+        run: |
+          pip install -r bal_tools/requirements-dev.txt
+          pytest

.gitignore

@@ -4,3 +4,5 @@ __pycache__/
 .env
 Pipfile*
 .pytest_cache
+
+tests/payload_outputs/*.json

bal_tools/safe_tx_builder/__init__.py

@@ -0,0 +1,3 @@
+from .safe_tx_builder import SafeTxBuilder
+from .safe_contract import SafeContract
+from .abi import ContractABI, ABIFunction

bal_tools/safe_tx_builder/abi.py

@@ -0,0 +1,63 @@
+from dataclasses import dataclass
+from typing import List, Optional
+
+
+@dataclass
+class InputType:
+    name: str
+    type: str
+
+
+@dataclass
+class ABIFunction:
+    name: Optional[str] = None
+    inputs: Optional[List[InputType]] = None
+    outputs: List[str] = None
+    constant: bool = False
+    payable: bool = False
+
+
+@dataclass
+class ContractABI:
+    functions: List[ABIFunction]
+    name: Optional[str] = None
+
+
+def parse_json_abi(abi: dict) -> ContractABI:
+    functions = []
+    for entry in abi:
+        if entry["type"] == "function":
+            name = entry["name"]
+            input_types = [collapse_if_tuple(inputs) for inputs in entry["inputs"]]
+            input_names = [i["name"] for i in entry["inputs"]]
+            inputs = [
+                InputType(name=name, type=typ)
+                for name, typ in zip(input_names, input_types)
+            ]
+            outputs = [collapse_if_tuple(outputs) for outputs in entry["outputs"]]
+            constant = entry["stateMutability"] in ("view", "pure")
+            payable = entry["stateMutability"] == "payable"
+            functions.append(
+                ABIFunction(
+                    name=name,
+                    inputs=inputs,
+                    outputs=outputs,
+                    constant=constant,
+                    payable=payable,
+                )
+            )
+    return ContractABI(functions)
+
+
+def collapse_if_tuple(abi: dict) -> str:
+    typ = abi["type"]
+    if not typ.startswith("tuple"):
+        return typ
+
+    delimited = ",".join(collapse_if_tuple(c) for c in abi["components"])
+    # Whatever comes after "tuple" is the array dims.  The ABI spec states that
+    # this will have the form "", "[]", or "[k]".
+    array_dim = typ[5:]
+    collapsed = "({}){}".format(delimited, array_dim)
+
+    return collapsed

bal_tools/safe_tx_builder/models.py

@@ -0,0 +1,57 @@
+from pydantic import BaseModel, Field, root_validator
+from typing import List, Optional, Union
+from datetime import datetime
+from enum import Enum
+
+
+class Meta(BaseModel):
+    name: str = Field(default="Transactions Batch")
+    description: str = Field(default="")
+    txBuilderVersion: str = Field(default="1.16.3")
+    createdFromSafeAddress: str = Field(default="")
+    createdFromOwnerAddress: str = Field(default="")
+    checksum: str = Field(
+        default="0x0000000000000000000000000000000000000000000000000000000000000000"
+    )
+
+
+class ContractMethod(BaseModel):
+    inputs: List[dict] = Field(default_factory=list)
+    name: str = Field(default="")
+    payable: bool = Field(default=False)
+
+
+class Transaction(BaseModel):
+    to: str = Field(default="")
+    value: str = Field(default="0")
+    data: Optional[str] = Field(default=None)
+    contractMethod: ContractMethod = Field(default_factory=ContractMethod)
+    contractInputsValues: dict = Field(default_factory=dict)
+
+
+class InputType(BaseModel):
+    name: str = Field(default="")
+    type: str = Field(default="")
+    internalType: str = Field(default="")
+
+
+class BasePayload(BaseModel):
+    version: str = Field(default="1.0")
+    chainId: str = Field(default="1")
+    createdAt: int = Field(default_factory=lambda: int(datetime.utcnow().timestamp()))
+    meta: Meta = Field(default_factory=Meta)
+    transactions: List[Transaction] = Field(default_factory=list)
+
+
+class TemplateType(Enum):
+    BASE = ("base.json", BasePayload)
+    TRANSACTION = ("tx.json", Transaction)
+    INPUT_TYPE = ("input_type.json", InputType)
+
+    @property
+    def file_name(self):
+        return self.value[0]
+
+    @property
+    def model(self):
+        return self.value[1]

bal_tools/safe_tx_builder/safe_contract.py

@@ -0,0 +1,76 @@
+import json
+
+from .safe_tx_builder import SafeTxBuilder
+from .abi import ABIFunction, ContractABI, parse_json_abi
+from .models import *
+
+
+ZERO_ADDRESS = "0x0000000000000000000000000000000000000000"
+
+
+class SafeContract:
+    def __init__(
+        self,
+        address: str,
+        abi: dict = None,
+        abi_file_path: str = None,
+    ):
+        self.tx_builder = SafeTxBuilder()
+        self.address = self.tx_builder._resolve_address(address)
+        self.abi = self._load_abi(abi, abi_file_path)
+
+    def __getattr__(self, attribute):
+        if self.abi and hasattr(self.abi, "functions"):
+            for func in self.abi.functions:
+                if func.name == attribute:
+                    return lambda *args, **kwargs: self.call_function(
+                        func, args, kwargs
+                    )
+        raise AttributeError(f"No function named {attribute} in contract ABI")
+
+    def _load_abi(self, abi: dict = None, file_path: dict = None) -> ContractABI:
+        if not abi and not file_path:
+            raise ValueError("Either `abi` or `abi_file_path` must be provided")
+
+        if file_path:
+            with open(file_path, "r") as file:
+                abi = json.load(file)
+
+        return parse_json_abi(abi)
+
+    def _handle_type(self, value):
+        try:
+            if isinstance(value, float):
+                value = int(value)
+            return str(value)
+        except Exception as e:
+            raise ValueError(f"Failed to convert value to string: {e}")
+
+    def call_function(self, func: ABIFunction, args: tuple, kwargs: dict = {}):
+        if func.constant:
+            raise ValueError("Cannot build a tx for a constant function")
+
+        if len(args) != len(func.inputs):
+            raise ValueError("Number of arguments does not match function inputs")
+
+        tx = self.tx_builder.load_template(TemplateType.TRANSACTION)
+        tx.to = self.address
+        tx.contractMethod.name = func.name
+        tx.contractMethod.payable = func.payable
+        tx.value = str(kwargs.get("value", "0"))
+
+        if not func.inputs:
+            tx.contractInputsValues = None
+
+        for arg, input_type in zip(args, func.inputs):
+            if input_type.type == "address":
+                arg = self.tx_builder._resolve_address(arg)
+
+            input_template = self.tx_builder.load_template(TemplateType.INPUT_TYPE)
+            input_template.name = input_type.name
+            input_template.type = input_type.type
+            input_template.internalType = input_type.type
+            tx.contractMethod.inputs.append(input_template)
+            tx.contractInputsValues[input_type.name] = self._handle_type(arg)
+
+        self.tx_builder.base_payload.transactions.append(tx)

bal_tools/safe_tx_builder/safe_tx_builder.py

@@ -0,0 +1,84 @@
+from typing import Optional, Union
+from datetime import datetime
+import os
+
+from bal_addresses import AddrBook
+from web3 import Web3
+
+from .models import *
+
+
+class SafeTxBuilder:
+    _instance = None
+
+    def __new__(
+        cls,
+        safe_address: Optional[str] = None,
+        chain_id: str = "1",
+        version: str = "1.0",
+        timestamp: Optional[str] = None,
+        tx_builder_version: str = "1.16.3",
+    ):
+        if cls._instance is None:
+            if safe_address is None:
+                raise ValueError("`safe_address` is required")
+            cls._instance = super(SafeTxBuilder, cls).__new__(cls)
+            cls._instance._initialize(
+                safe_address, chain_id, version, timestamp, tx_builder_version
+            )
+        return cls._instance
+
+    def _initialize(
+        self,
+        safe_address: str,
+        chain_id: str,
+        version: str,
+        timestamp: Optional[str],
+        tx_builder_version: str,
+    ):
+        self.chain_id = chain_id
+        self.addr_book = AddrBook(
+            AddrBook.chain_names_by_id[int(self.chain_id)]
+        ).flatbook
+        self.safe_address = self._resolve_address(safe_address)
+        self.version = version
+        self.timestamp = timestamp if timestamp else datetime.utcnow().timestamp()
+        self.tx_builder_version = tx_builder_version
+        self.base_payload = self.load_template(TemplateType.BASE)
+        self._load_payload_metadata()
+
+    @staticmethod
+    def load_template(
+        template_type: TemplateType,
+    ) -> Union[BasePayload, Transaction, InputType]:
+        current_dir = os.path.dirname(os.path.abspath(__file__))
+        file_path = os.path.join(current_dir, "templates", template_type.file_name)
+
+        model = template_type.model
+        with open(file_path, "r") as f:
+            file_content = f.read()
+
+        return model.model_validate_json(file_content)
+
+    def _resolve_address(self, identifier: str) -> str:
+        if Web3.is_address(identifier):
+            return identifier
+
+        return self.addr_book[identifier]
+
+    def _load_payload_metadata(self):
+        self.base_payload.version = self.version
+        self.base_payload.chainId = self.chain_id
+        self.base_payload.createdAt = int(self.timestamp)
+        self.base_payload.meta.txBuilderVersion = self.tx_builder_version
+        self.base_payload.meta.createdFromSafeAddress = self.safe_address
+
+    def output_payload(self, output_file: str) -> BasePayload:
+        """
+        output the final json payload to `output_file`
+        returns the payload
+        """
+        with open(output_file, "w") as f:
+            f.write(self.base_payload.model_dump_json())
+
+        return self.base_payload

bal_tools/safe_tx_builder/templates/base.json

@@ -0,0 +1,14 @@
+{
+    "version": "",
+    "chainId": "",
+    "createdAt": 0,
+    "meta": {
+        "name": "Transactions Batch",
+        "description": "",
+        "txBuilderVersion": "",
+        "createdFromSafeAddress": "",
+        "createdFromOwnerAddress": "",
+        "checksum": "0x0000000000000000000000000000000000000000000000000000000000000000"
+    },
+    "transactions": []
+}

bal_tools/safe_tx_builder/templates/input_type.json

@@ -0,0 +1,5 @@
+{
+    "name": "",
+    "type": "",
+    "internalType": ""
+}

bal_tools/safe_tx_builder/templates/tx.json

@@ -0,0 +1,11 @@
+{
+    "to": "",
+    "value": "0",
+    "data": null,
+    "contractMethod": {
+      "inputs": [],
+      "name": "",
+      "payable": false
+    },
+    "contractInputsValues": {}
+  }