feat(BACK-7982): add erc7730 list/erc7730 format commands

docs/pages/usage_cli.md

@@ -30,6 +30,20 @@ You can validate your setup by running the `erc7730` command:
 
 ## Commands
 
+### `erc7730 list`
+
+The `list` command recursively lists descriptors files in directory:
+
+```shell
+$ erc7730 list
+ercs/calldata-erc721-nfts.json
+ercs/eip712-erc2612-permit.json
+ercs/calldata-erc20-tokens.json
+registry/1inch/calldata-AggregationRouterV5.json
+registry/1inch/eip712-1inch-aggregation-router.json
+...
+```
+
 ### `erc7730 lint`
 
 The `lint` command runs validations on descriptors and outputs warnings and errors to the console:
@@ -111,3 +125,22 @@ $ erc7730 schema                    # print JSON schema of input form (ERC-7730
 $ erc7730 schema resolved           # print JSON schema of resolved form
 $ erc7730 resolve <descriptor>.json # convert descriptor from input to resolved form
 ```
+
+### `erc7730 format`
+
+The `format` command recursively finds and formats all descriptor files, starting from current directory by default:
+
+```shell
+$ erc7730 format
+📝 formatting 294 descriptor files…
+
+➡️ formatting registry/uniswap/eip712-uniswap-permit2.json…
+no issue found ✔️
+
+➡️ formatting registry/tether/calldata-usdt.json…
+no issue found ✔️
+
+...
+
+formatted 294 descriptor files, no errors occurred ✅
+```

src/erc7730/common/client.py

@@ -62,6 +62,8 @@ def get_contract_abis(chain_id: int, contract_address: Address) -> list[ABI]:
     except Exception as e:
         if "Contract source code not verified" in str(e):
             raise Exception("contract source is not available on Etherscan") from e
+        if "Max calls per sec rate limit reached" in str(e):
+            raise Exception("Etherscan rate limit exceeded, please retry") from e
         raise e
 
 

src/erc7730/common/json.py

@@ -1,4 +1,5 @@
 import json
+import os
 from collections.abc import Iterator
 from json import JSONEncoder
 from pathlib import Path
@@ -35,9 +36,7 @@ def read_json_with_includes(path: Path) -> Any:
       - circular includes are not detected and will result in a stack overflow.
       - "includes" key can only be used at root level of an object.
     """
-    result: Any
-    with open(path) as f:
-        result = json.load(f)
+    result: dict[str, Any] = dict_from_json_file(path)
     if isinstance(result, dict) and (includes := result.pop("includes", None)) is not None:
         if isinstance(includes, list):
             parent = read_jsons_with_includes(paths=[path.parent / p for p in includes])
@@ -68,6 +67,30 @@ def _merge_dicts(d1: dict[str, Any], d2: dict[str, Any]) -> dict[str, Any]:
     return {**d2, **merged}
 
 
+def dict_from_json_str(value: str) -> dict[str, Any]:
+    """Deserialize a dict from a JSON string."""
+    return json.loads(value)
+
+
+def dict_from_json_file(path: Path) -> dict[str, Any]:
+    """Deserialize a dict from a JSON file."""
+    with open(path, "rb") as f:
+        return json.load(f)
+
+
+def dict_to_json_str(values: dict[str, Any]) -> str:
+    """Serialize a dict into a JSON string."""
+    return json.dumps(values, indent=2, cls=CompactJSONEncoder)
+
+
+def dict_to_json_file(path: Path, values: dict[str, Any]) -> None:
+    """Serialize a dict into a JSON file, creating parent directories as needed."""
+    os.makedirs(path.parent, exist_ok=True)
+    with open(path, "w") as f:
+        f.write(dict_to_json_str(values))
+        f.write("\n")
+
+
 class CompactJSONEncoder(JSONEncoder):
     """A JSON Encoder that puts small containers on single lines."""
 

src/erc7730/common/pydantic.py

@@ -1,4 +1,3 @@
-import json
 import os
 from collections.abc import Callable
 from dataclasses import dataclass
@@ -8,7 +7,7 @@
 from pydantic import BaseModel, ValidationInfo, WrapValidator
 from pydantic_core import PydanticCustomError
 
-from erc7730.common.json import CompactJSONEncoder, read_json_with_includes
+from erc7730.common.json import dict_to_json_file, dict_to_json_str, read_json_with_includes
 
 _BaseModel = TypeVar("_BaseModel", bound=BaseModel)
 
@@ -40,15 +39,12 @@ def model_to_json_dict(obj: _BaseModel) -> dict[str, Any]:
 
 def model_to_json_str(obj: _BaseModel) -> str:
     """Serialize a pydantic model into a JSON string."""
-    return json.dumps(model_to_json_dict(obj), indent=2, cls=CompactJSONEncoder)
+    return dict_to_json_str(model_to_json_dict(obj))
 
 
 def model_to_json_file(path: Path, model: _BaseModel) -> None:
     """Write a model to a JSON file, creating parent directories as needed."""
-    os.makedirs(path.parent, exist_ok=True)
-    with open(path, "w") as f:
-        f.write(model_to_json_str(model))
-        f.write("\n")
+    dict_to_json_file(path, model_to_json_dict(model))
 
 
 @dataclass(frozen=True)

src/erc7730/format/__init__.py

@@ -0,0 +1 @@
+"""Package implementing formatting commands to normalize descriptor files."""

src/erc7730/format/format.py

@@ -0,0 +1,83 @@
+import os
+from concurrent.futures.thread import ThreadPoolExecutor
+from pathlib import Path
+
+from rich import print
+
+from erc7730.common.json import dict_from_json_file, dict_to_json_file
+from erc7730.common.output import (
+    AddFileOutputAdder,
+    BufferAdder,
+    ConsoleOutputAdder,
+    DropFileOutputAdder,
+    ExceptionsToOutput,
+    OutputAdder,
+)
+from erc7730.list.list import get_erc7730_files
+
+
+def format_all_and_print_errors(paths: list[Path]) -> bool:
+    """
+    Format all ERC-7730 descriptor files at given paths and print errors.
+
+    :param paths: paths to apply formatter on
+    :return: true if not errors occurred
+    """
+    out = DropFileOutputAdder(delegate=ConsoleOutputAdder())
+
+    count = format_all(paths, out)
+
+    if out.has_errors:
+        print(f"[bold][red]formatted {count} descriptor files, some errors occurred ❌[/red][/bold]")
+        return False
+
+    if out.has_warnings:
+        print(f"[bold][yellow]formatted {count} descriptor files, some warnings occurred ⚠️[/yellow][/bold]")
+        return True
+
+    print(f"[bold][green]formatted {count} descriptor files, no errors occurred ✅[/green][/bold]")
+    return True
+
+
+def format_all(paths: list[Path], out: OutputAdder) -> int:
+    """
+    Format all ERC-7730 descriptor files at given paths.
+
+    Paths can be files or directories, in which case all descriptor files in the directory are recursively formatted.
+
+    :param paths: paths to apply formatter on
+    :param out: output adder
+    :return: number of files formatted
+    """
+    files = list(get_erc7730_files(*paths, out=out))
+
+    if len(files) <= 1 or not (root_path := os.path.commonpath(files)):
+        root_path = None
+
+    def label(f: Path) -> Path | None:
+        return f.relative_to(root_path) if root_path is not None else None
+
+    if len(files) > 1:
+        print(f"📝 formatting {len(files)} descriptor files…\n")
+
+    with ThreadPoolExecutor() as executor:
+        for future in (executor.submit(format_file, file, out, label(file)) for file in files):
+            future.result()
+
+    return len(files)
+
+
+def format_file(path: Path, out: OutputAdder, show_as: Path | None = None) -> None:
+    """
+    Format a single ERC-7730 descriptor file.
+
+    :param path: ERC-7730 descriptor file path
+    :param show_as: if provided, print this label instead of the file path
+    :param out: error handler
+    """
+
+    label = path if show_as is None else show_as
+    file_out = AddFileOutputAdder(delegate=out, file=path)
+
+    with BufferAdder(file_out, prolog=f"➡️ formatting [bold]{label}[/bold]…", epilog="") as out, ExceptionsToOutput(out):
+        dict_to_json_file(path, dict_from_json_file(path))

src/erc7730/lint/lint.py

@@ -1,11 +1,9 @@
 import os
-from collections.abc import Generator
 from concurrent.futures.thread import ThreadPoolExecutor
 from pathlib import Path
 
 from rich import print
 
-from erc7730 import ERC_7730_REGISTRY_CALLDATA_PREFIX, ERC_7730_REGISTRY_EIP712_PREFIX
 from erc7730.common.output import (
     AddFileOutputAdder,
     BufferAdder,
@@ -21,6 +19,7 @@
 from erc7730.lint.lint_transaction_type_classifier import ClassifyTransactionTypeLinter
 from erc7730.lint.lint_validate_abi import ValidateABILinter
 from erc7730.lint.lint_validate_display_fields import ValidateDisplayFieldsLinter
+from erc7730.list.list import get_erc7730_files
 from erc7730.model.input.descriptor import InputERC7730Descriptor
 
 
@@ -59,20 +58,7 @@ def lint_all(paths: list[Path], out: OutputAdder) -> int:
         ]
     )
 
-    def get_descriptor_files() -> Generator[Path, None, None]:
-        for path in paths:
-            if path.is_file():
-                yield path
-            elif path.is_dir():
-                for file in path.rglob("*.json"):
-                    if file.name.startswith(ERC_7730_REGISTRY_CALLDATA_PREFIX) or file.name.startswith(
-                        ERC_7730_REGISTRY_EIP712_PREFIX
-                    ):
-                        yield file
-            else:
-                raise ValueError(f"Invalid path: {path}")
-
-    files = list(get_descriptor_files())
+    files = list(get_erc7730_files(*paths, out=out))
 
     if len(files) <= 1 or not (root_path := os.path.commonpath(files)):
         root_path = None

src/erc7730/list/__init__.py

@@ -0,0 +1 @@
+"""Package implementing listing commands to easily find descriptor files."""

src/erc7730/list/list.py

@@ -0,0 +1,62 @@
+from collections.abc import Generator
+from pathlib import Path
+
+from rich import print
+
+from erc7730 import ERC_7730_REGISTRY_CALLDATA_PREFIX, ERC_7730_REGISTRY_EIP712_PREFIX
+from erc7730.common.output import (
+    ConsoleOutputAdder,
+    OutputAdder,
+)
+
+
+def list_all(paths: list[Path]) -> bool:
+    """
+    List all ERC-7730 descriptor files at given paths.
+
+    Paths can be files or directories, in which case all descriptor files in the directory are recursively listed.
+
+    :param paths: paths to search for descriptor files
+    :return: true if no error occurred
+    """
+    out = ConsoleOutputAdder()
+
+    for file in get_erc7730_files(*paths, out=out):
+        print(file)
+
+    return not out.has_errors
+
+
+def get_erc7730_files(*paths: Path, out: OutputAdder) -> Generator[Path, None, None]:
+    """
+    List all ERC-7730 descriptor files at given paths.
+
+    Paths can be files or directories, in which case all descriptor files in the directory are recursively listed.
+
+    :param paths: paths to search for descriptor files
+    :param out: error handler
+    """
+    for path in paths:
+        if path.is_file():
+            if is_erc7730_file(path):
+                yield path
+            else:
+                out.error(title="Invalid path", message=f"{path} is not an ERC-7730 descriptor file")
+        elif path.is_dir():
+            for file in path.rglob("*.json"):
+                if is_erc7730_file(file):
+                    yield file
+        else:
+            out.error(title="Invalid path", message=f"{path} is not a file or directory")
+
+
+def is_erc7730_file(path: Path) -> bool:
+    """
+    Check if a file is an ERC-7730 descriptor file.
+
+    :param path: file path
+    :return: true if the file is an ERC-7730 descriptor file
+    """
+    return path.is_file() and (
+        path.name.startswith(ERC_7730_REGISTRY_CALLDATA_PREFIX) or path.name.startswith(ERC_7730_REGISTRY_EIP712_PREFIX)
+    )