add
This commit is contained in:
lz_db
220
ccxt/static_dependencies/starknet/abi/v1/parser.py
Normal file
220
ccxt/static_dependencies/starknet/abi/v1/parser.py
Normal file
@@ -0,0 +1,220 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import json
|
||||
import os
|
||||
from collections import OrderedDict, defaultdict
|
||||
from pathlib import Path
|
||||
from typing import DefaultDict, Dict, List, Optional, Tuple, Union, cast
|
||||
|
||||
from ....marshmallow import EXCLUDE
|
||||
|
||||
from .model import Abi
|
||||
from .schemas import ContractAbiEntrySchema
|
||||
from .shape import (
|
||||
ENUM_ENTRY,
|
||||
EVENT_ENTRY,
|
||||
FUNCTION_ENTRY,
|
||||
STRUCT_ENTRY,
|
||||
EventDict,
|
||||
FunctionDict,
|
||||
TypedParameterDict,
|
||||
)
|
||||
from ...cairo.data_types import CairoType, EnumType, StructType
|
||||
from ...cairo.v1.type_parser import TypeParser
|
||||
|
||||
|
||||
class AbiParsingError(ValueError):
|
||||
"""
|
||||
Error raised when something wrong goes during abi parsing.
|
||||
"""
|
||||
|
||||
|
||||
class AbiParser:
|
||||
"""
|
||||
Utility class for parsing abi into a dataclass.
|
||||
"""
|
||||
|
||||
# Entries from ABI grouped by entry type
|
||||
_grouped: DefaultDict[str, List[Dict]]
|
||||
# lazy init property
|
||||
_type_parser: Optional[TypeParser] = None
|
||||
|
||||
def __init__(self, abi_list: List[Dict]):
|
||||
"""
|
||||
Abi parser constructor. Ensures that abi satisfies the abi schema.
|
||||
|
||||
:param abi_list: Contract's ABI as a list of dictionaries.
|
||||
"""
|
||||
# prepend abi with core structures
|
||||
core_structures = (
|
||||
Path(os.path.dirname(__file__)) / "core_structures.json"
|
||||
).read_text("utf-8")
|
||||
abi_list = json.loads(core_structures)["abi"] + abi_list
|
||||
abi = [
|
||||
ContractAbiEntrySchema().load(entry, unknown=EXCLUDE) for entry in abi_list
|
||||
]
|
||||
grouped = defaultdict(list)
|
||||
for entry in abi:
|
||||
assert isinstance(entry, dict)
|
||||
grouped[entry["type"]].append(entry)
|
||||
|
||||
self._grouped = grouped
|
||||
|
||||
def parse(self) -> Abi:
|
||||
"""
|
||||
Parse abi provided to constructor and return it as a dataclass. Ensures that there are no cycles in the abi.
|
||||
|
||||
:raises: AbiParsingError: on any parsing error.
|
||||
:return: Abi dataclass.
|
||||
"""
|
||||
structures, enums = self._parse_structures_and_enums()
|
||||
functions_dict = cast(
|
||||
Dict[str, FunctionDict],
|
||||
AbiParser._group_by_entry_name(
|
||||
self._grouped[FUNCTION_ENTRY], "defined functions"
|
||||
),
|
||||
)
|
||||
events_dict = cast(
|
||||
Dict[str, EventDict],
|
||||
AbiParser._group_by_entry_name(
|
||||
self._grouped[EVENT_ENTRY], "defined events"
|
||||
),
|
||||
)
|
||||
|
||||
return Abi(
|
||||
defined_structures=structures,
|
||||
defined_enums=enums,
|
||||
functions={
|
||||
name: self._parse_function(entry)
|
||||
for name, entry in functions_dict.items()
|
||||
},
|
||||
events={
|
||||
name: self._parse_event(entry) for name, entry in events_dict.items()
|
||||
},
|
||||
)
|
||||
|
||||
@property
|
||||
def type_parser(self) -> TypeParser:
|
||||
if self._type_parser:
|
||||
return self._type_parser
|
||||
|
||||
raise RuntimeError("Tried to get type_parser before it was set.")
|
||||
|
||||
def _parse_structures_and_enums(
|
||||
self,
|
||||
) -> Tuple[Dict[str, StructType], Dict[str, EnumType]]:
|
||||
structs_dict = AbiParser._group_by_entry_name(
|
||||
self._grouped[STRUCT_ENTRY], "defined structures"
|
||||
)
|
||||
enums_dict = AbiParser._group_by_entry_name(
|
||||
self._grouped[ENUM_ENTRY], "defined enums"
|
||||
)
|
||||
|
||||
# Contains sorted members of the struct
|
||||
struct_members: Dict[str, List[TypedParameterDict]] = {}
|
||||
structs: Dict[str, StructType] = {}
|
||||
|
||||
# Contains sorted members of the enum
|
||||
enum_members: Dict[str, List[TypedParameterDict]] = {}
|
||||
enums: Dict[str, EnumType] = {}
|
||||
|
||||
# Example problem (with a simplified json structure):
|
||||
# [{name: User, fields: {id: Uint256}}, {name: "Uint256", ...}]
|
||||
# User refers to Uint256 even though it is not known yet (will be parsed next).
|
||||
# This is why it is important to create the structure types first. This way other types can already refer to
|
||||
# them when parsing types, even thought their fields are not filled yet.
|
||||
# At the end we will mutate those structures to contain the right fields. An alternative would be to use
|
||||
# topological sorting with an additional "unresolved type", so this flow is much easier.
|
||||
for name, struct in structs_dict.items():
|
||||
structs[name] = StructType(name, OrderedDict())
|
||||
struct_members[name] = struct["members"]
|
||||
|
||||
for name, enum in enums_dict.items():
|
||||
enums[name] = EnumType(name, OrderedDict())
|
||||
enum_members[name] = enum["variants"]
|
||||
|
||||
# Now parse the types of members and save them.
|
||||
defined_structs_enums: Dict[str, Union[StructType, EnumType]] = dict(structs)
|
||||
defined_structs_enums.update(enums)
|
||||
|
||||
self._type_parser = TypeParser(defined_structs_enums)
|
||||
for name, struct in structs.items():
|
||||
members = self._parse_members(
|
||||
cast(List[TypedParameterDict], struct_members[name]),
|
||||
f"members of structure '{name}'",
|
||||
)
|
||||
struct.types.update(members)
|
||||
for name, enum in enums.items():
|
||||
members = self._parse_members(
|
||||
cast(List[TypedParameterDict], enum_members[name]),
|
||||
f"members of enum '{name}'",
|
||||
)
|
||||
enum.variants.update(members)
|
||||
|
||||
# All types have their members assigned now
|
||||
|
||||
self._check_for_cycles(defined_structs_enums)
|
||||
|
||||
return structs, enums
|
||||
|
||||
@staticmethod
|
||||
def _check_for_cycles(structs: Dict[str, Union[StructType, EnumType]]):
|
||||
# We want to avoid creating our own cycle checker as it would make it more complex. json module has a built-in
|
||||
# checker for cycles.
|
||||
try:
|
||||
_to_json(structs)
|
||||
except ValueError as err:
|
||||
raise AbiParsingError(err) from ValueError
|
||||
|
||||
def _parse_function(self, function: FunctionDict) -> Abi.Function:
|
||||
return Abi.Function(
|
||||
name=function["name"],
|
||||
inputs=self._parse_members(function["inputs"], function["name"]),
|
||||
outputs=list(
|
||||
self.type_parser.parse_inline_type(param["type"])
|
||||
for param in function["outputs"]
|
||||
),
|
||||
)
|
||||
|
||||
def _parse_event(self, event: EventDict) -> Abi.Event:
|
||||
return Abi.Event(
|
||||
name=event["name"],
|
||||
inputs=self._parse_members(event["inputs"], event["name"]),
|
||||
)
|
||||
|
||||
def _parse_members(
|
||||
self, params: List[TypedParameterDict], entity_name: str
|
||||
) -> OrderedDict[str, CairoType]:
|
||||
# Without cast, it complains that 'Type "TypedParameterDict" cannot be assigned to type "T@_group_by_name"'
|
||||
members = AbiParser._group_by_entry_name(cast(List[Dict], params), entity_name)
|
||||
return OrderedDict(
|
||||
(name, self.type_parser.parse_inline_type(param["type"]))
|
||||
for name, param in members.items()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _group_by_entry_name(
|
||||
dicts: List[Dict], entity_name: str
|
||||
) -> OrderedDict[str, Dict]:
|
||||
grouped = OrderedDict()
|
||||
for entry in dicts:
|
||||
name = entry["name"]
|
||||
if name in grouped:
|
||||
raise AbiParsingError(
|
||||
f"Name '{name}' was used more than once in {entity_name}."
|
||||
)
|
||||
grouped[name] = entry
|
||||
return grouped
|
||||
|
||||
|
||||
def _to_json(value):
|
||||
class DataclassSupportingEncoder(json.JSONEncoder):
|
||||
def default(self, o):
|
||||
# Dataclasses are not supported by json. Additionally, dataclasses.asdict() works recursively and doesn't
|
||||
# check for cycles, so we need to flatten dataclasses (by ONE LEVEL) ourselves.
|
||||
if dataclasses.is_dataclass(o):
|
||||
return tuple(getattr(o, field.name) for field in dataclasses.fields(o))
|
||||
return super().default(o)
|
||||
|
||||
return json.dumps(value, cls=DataclassSupportingEncoder)
|
||||
Reference in New Issue
Block a user