abi_resolver.py•11 kB
"""ABI Resolver - Using Python data objects for clean manipulation"""
import json
import logging
from typing import Any, Dict, Optional, Set
from .abi_types import (
ABI, ABIStruct, ABIVariant, ABIType, ABIProtobufType
)
logger = logging.getLogger(__name__)
class ABIResolver:
"""Resolves ABI types using Python data objects"""
def __init__(self, abi_path: Optional[str] = None, abi_data: Optional[Dict] = None):
"""
Initialize ABI Resolver
Args:
abi_path: Path to ABI JSON file
abi_data: ABI dictionary (if already loaded)
"""
if abi_data:
self.abi = ABI.from_json(abi_data)
elif abi_path:
with open(abi_path, 'r', encoding='utf-8') as f:
self.abi = ABI.from_json(json.load(f))
else:
raise ValueError("Either abi_path or abi_data must be provided")
def resolve_action(self, action_name: str) -> Dict[str, Any]:
"""
Resolve an action to its complete JSON template
Args:
action_name: Name of the action
Returns:
Complete JSON template with examples and hints
"""
action = self.abi.get_action(action_name)
if not action:
raise ValueError(f"Action '{action_name}' not found in ABI")
# Resolve the type
resolved = self.resolve_type(action.type)
return {
"action": action_name,
"type": action.type,
"description": f"Parameters for {action_name} action",
"template": resolved["template"],
"schema": resolved["schema"],
"example": self._fill_example(resolved["template"])
}
def resolve_type(self, type_name: str, visited: Optional[Set[str]] = None) -> Dict[str, Any]:
"""
Recursively resolve a type by traversing the ABI DAG
Args:
type_name: Type name to resolve
visited: Set of already visited types (for cycle detection)
Returns:
Dictionary with 'template' and 'schema' keys
"""
if visited is None:
visited = set()
# Handle arrays
if type_name.endswith("[]"):
base_type = type_name[:-2]
resolved = self.resolve_type(base_type, visited)
return {
"template": [resolved["template"]] if resolved["template"] else [],
"schema": {
"type": "array",
"items": resolved["schema"]
}
}
# Handle optional types
if type_name.endswith("?"):
base_type = type_name[:-1]
resolved = self.resolve_type(base_type, visited)
schema = resolved["schema"].copy()
schema["nullable"] = True
return {
"template": resolved["template"],
"schema": schema
}
# Check for cycles
if type_name in visited:
return {
"template": {"$ref": type_name},
"schema": {"$ref": f"#/definitions/{type_name}"}
}
visited.add(type_name)
# Check if it's a base type
if self._is_base_type(type_name):
return self._resolve_base_type(type_name)
# Get type definition from ABI
type_def = self.abi.get_type_definition(type_name)
if isinstance(type_def, ABIStruct):
return self._resolve_struct(type_def, visited)
elif isinstance(type_def, ABIVariant):
return self._resolve_variant(type_def, visited)
elif isinstance(type_def, ABIType):
# Type alias - resolve the underlying type
return self.resolve_type(type_def.type, visited)
elif isinstance(type_def, ABIProtobufType):
return self._resolve_protobuf_type(type_def, visited)
else:
# Unknown type
logger.warning(f"Unknown type: {type_name}")
return {
"template": {},
"schema": {"type": "object", "additionalProperties": True}
}
def _is_base_type(self, type_name: str) -> bool:
"""Check if type is a base type"""
base_types = {
"string", "bool", "bytes",
"int8", "uint8", "int16", "uint16",
"int32", "uint32", "int64", "uint64",
"int128", "uint128", "float32", "float64",
"checksum256", "checksum160", "checksum512",
"name", "asset", "symbol", "public_key", "signature"
}
return type_name in base_types
def _resolve_base_type(self, type_name: str) -> Dict[str, Any]:
"""Resolve a base type"""
if type_name == "string":
return {"template": "", "schema": {"type": "string"}}
elif type_name == "bool":
return {"template": False, "schema": {"type": "boolean"}}
elif type_name.startswith(("int", "uint")):
return {"template": 0, "schema": {"type": "integer", "format": type_name}}
elif type_name.startswith("float"):
return {"template": 0.0, "schema": {"type": "number", "format": type_name}}
elif type_name.startswith("checksum"):
size = int(type_name[8:]) // 4 # Convert bits to hex chars
return {
"template": "0x" + "0" * size,
"schema": {
"type": "string",
"pattern": f"^0x[a-fA-F0-9]{{{size}}}$"
}
}
else:
return {"template": "", "schema": {"type": "string", "format": type_name}}
def _resolve_struct(self, struct: ABIStruct, visited: Set[str]) -> Dict[str, Any]:
"""Resolve a struct type"""
if struct.is_empty():
return {
"template": {},
"schema": {
"type": "object",
"properties": {},
"description": f"Empty struct: {struct.name}"
}
}
template = {}
properties = {}
required = []
for field in struct.fields:
# Resolve field type
resolved = self.resolve_type(field.type, visited.copy())
template[field.name] = resolved["template"]
properties[field.name] = resolved["schema"]
# Add to required unless optional
if not field.is_optional:
required.append(field.name)
schema = {
"type": "object",
"properties": properties
}
if required:
schema["required"] = required
return {"template": template, "schema": schema}
def _resolve_variant(self, variant: ABIVariant, visited: Set[str]) -> Dict[str, Any]:
"""Resolve a variant (union) type"""
if not variant.types:
return {"template": {}, "schema": {"type": "object"}}
# Use first type as template
first_resolved = self.resolve_type(variant.types[0], visited.copy())
# Build oneOf schema
one_of = []
for var_type in variant.types:
resolved = self.resolve_type(var_type, visited.copy())
one_of.append(resolved["schema"])
return {
"template": first_resolved["template"],
"schema": {
"oneOf": one_of,
"description": f"Variant: {variant.name}"
}
}
def _resolve_protobuf_type(self, pb_type: ABIProtobufType, visited: Set[str]) -> Dict[str, Any]:
"""Resolve a protobuf type"""
template = {}
properties = {}
required = []
for field_name, field_info in pb_type.fields.items():
field_type = field_info.get("type", "string")
is_required = field_info.get("required", False)
if field_type in ["string", "object", "array", "uint64", "uint32"]:
# Handle basic types
if field_type == "string":
template[field_name] = ""
properties[field_name] = {"type": "string"}
elif field_type == "object":
template[field_name] = {}
properties[field_name] = {"type": "object"}
elif field_type == "array":
template[field_name] = []
properties[field_name] = {"type": "array"}
elif field_type.startswith("uint"):
template[field_name] = 0
properties[field_name] = {"type": "integer"}
else:
# Nested protobuf type
resolved = self.resolve_type(field_type, visited.copy())
template[field_name] = resolved["template"]
properties[field_name] = resolved["schema"]
if is_required:
required.append(field_name)
schema = {
"type": "object",
"properties": properties,
"description": f"Protobuf type: {pb_type.name}"
}
if required:
schema["required"] = required
return {"template": template, "schema": schema}
def _fill_example(self, template: Any) -> Any:
"""Fill template with example values"""
if isinstance(template, dict):
example = {}
for key, value in template.items():
if key == "$ref":
example[key] = value
elif value == "":
example[key] = f"example_{key}"
elif value == 0:
example[key] = 12345
elif value == 0.0:
example[key] = 123.45
elif value is None:
example[key] = None
else:
example[key] = self._fill_example(value)
return example
elif isinstance(template, list):
return [self._fill_example(template[0])] if template else []
else:
return template
def get_all_actions(self) -> list[str]:
"""Get list of all available actions"""
return list(self.abi.actions.keys())
def get_action_info(self, action_name: str) -> Dict[str, Any]:
"""Get comprehensive information about an action"""
try:
return self.resolve_action(action_name)
except Exception as e:
logger.error(f"Error resolving action {action_name}: {e}")
return {
"action": action_name,
"error": str(e),
"template": {},
"schema": {},
"example": {}
}