abi_fetcher.py•7.45 kB
"""ABI fetcher for NIX contracts using cleos"""
import json
import logging
import os
import subprocess
from typing import Dict, Any, Optional, List
from pathlib import Path
from .abi_resolver import ABIResolver
logger = logging.getLogger(__name__)
class ABIFetcher:
"""Fetch and parse contract ABIs using cleos"""
def __init__(self,
nodeos_api: str = None,
environment: str = None,
cleos_path: str = None,
cleos_docker_cmd: str = None,
cleos_custom_cmd: str = None):
"""
Initialize ABI fetcher with cleos configuration
Args:
nodeos_api: Nodeos API endpoint (overrides environment)
environment: Environment name (dev, uat, prod, etc.)
cleos_path: Path to cleos binary
cleos_docker_cmd: Docker command for cleos
cleos_custom_cmd: Custom command for cleos
"""
# If explicit endpoint is provided, use it
if nodeos_api:
self.nodeos_api = nodeos_api
# Otherwise, use environment configuration
else:
from .env_config import EnvironmentConfig
env_config = EnvironmentConfig()
env = environment or os.getenv("NODEOS_ENV", "cdev")
nodeos_endpoint, _ = env_config.get_endpoints(env)
# Use environment-specific endpoint, not os.getenv()
self.nodeos_api = nodeos_endpoint
# Determine cleos command
if cleos_custom_cmd:
self.cleos_cmd = cleos_custom_cmd.split()
elif cleos_docker_cmd:
self.cleos_cmd = cleos_docker_cmd.split()
elif cleos_path:
self.cleos_cmd = [cleos_path]
else:
# Try to find cleos in PATH or use environment variable
self.cleos_cmd = [os.getenv("CLEOS_PATH", "cleos")]
def _run_cleos(self, *args) -> str:
"""
Run a cleos command
Args:
*args: Command arguments
Returns:
Command output as string
Raises:
RuntimeError: If command fails
"""
cmd = self.cleos_cmd + ["-u", self.nodeos_api] + list(args)
logger.debug(f"Running command: {' '.join(cmd)}")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=True
)
return result.stdout
except subprocess.CalledProcessError as e:
logger.error(f"Cleos command failed: {e.stderr}")
raise RuntimeError(f"Failed to execute cleos: {e.stderr}")
except FileNotFoundError:
logger.error(f"Cleos not found. Please check your configuration.")
raise RuntimeError("Cleos binary not found. Please install cleos or configure the path correctly.")
def get_abi(self, account: str) -> Dict[str, Any]:
"""
Fetch ABI for a specific account
Args:
account: Account name (e.g., 'nix.q')
Returns:
ABI as dictionary
"""
logger.info(f"Fetching ABI for account: {account}")
try:
output = self._run_cleos("get", "abi", account)
# The ABI JSON is directly at the root level
abi = json.loads(output)
# Verify it's a valid ABI by checking for expected fields
if "version" not in abi and "actions" not in abi:
raise ValueError(f"Invalid ABI structure for account {account}")
return abi
except json.JSONDecodeError as e:
logger.error(f"Failed to parse ABI JSON: {e}")
raise ValueError(f"Invalid ABI response for {account}")
def get_actions(self, account: str) -> list:
"""
Get list of actions from contract ABI
Args:
account: Account name
Returns:
List of action names
"""
abi = self.get_abi(account)
return [action["name"] for action in abi.get("actions", [])]
def get_action_schema(self, account: str, action_name: str) -> Optional[Dict[str, Any]]:
"""
Get complete resolved schema for a specific action
Args:
account: Account name
action_name: Action name
Returns:
Complete action schema with resolved types
"""
abi = self.get_abi(account)
# Use ABIResolver to get complete structure
resolver = ABIResolver(abi_data=abi)
try:
return resolver.resolve_action(action_name)
except Exception as e:
logger.error(f"Failed to resolve action {action_name}: {e}")
return None
def get_action_template(self, account: str, action_name: str) -> Dict[str, Any]:
"""
Get a ready-to-use JSON template for an action
Args:
account: Account name
action_name: Action name
Returns:
JSON template with example values
"""
schema = self.get_action_schema(account, action_name)
if schema:
return schema.get("example", {})
return {}
def cache_abi(self, account: str, cache_dir: str = ".abi_cache"):
"""
Cache ABI to local file
Args:
account: Account name
cache_dir: Directory to store cached ABIs
"""
cache_path = Path(cache_dir)
cache_path.mkdir(exist_ok=True)
abi = self.get_abi(account)
cache_file = cache_path / f"{account}.json"
with open(cache_file, 'w') as f:
json.dump(abi, f, indent=2)
logger.info(f"Cached ABI for {account} to {cache_file}")
def load_cached_abi(self, account: str, cache_dir: str = ".abi_cache") -> Optional[Dict[str, Any]]:
"""
Load ABI from cache
Args:
account: Account name
cache_dir: Directory with cached ABIs
Returns:
Cached ABI or None if not found
"""
cache_file = Path(cache_dir) / f"{account}.json"
if cache_file.exists():
with open(cache_file, 'r') as f:
logger.info(f"Loaded cached ABI for {account}")
return json.load(f)
return None
if __name__ == "__main__":
# Example usage
import sys
logging.basicConfig(level=logging.INFO)
if len(sys.argv) > 1:
account = sys.argv[1]
else:
account = "nix.q"
fetcher = ABIFetcher()
try:
print(f"\nFetching ABI for {account}...")
abi = fetcher.get_abi(account)
print(f"\nActions available in {account}:")
actions = fetcher.get_actions(account)
for action in actions:
print(f" - {action}")
schema = fetcher.get_action_schema(account, action)
if schema:
print(f" Fields: {[field['name'] for field in schema.get('fields', [])]}")
# Cache the ABI
fetcher.cache_abi(account)
except Exception as e:
print(f"Error: {e}")
sys.exit(1)