simple_client.py•7.03 kB
"""Simplified NIX Client using JSON and cleos"""
import json
import logging
import os
import subprocess
from typing import Any, Dict, Union
from .env_config import EnvironmentConfig
logger = logging.getLogger(__name__)
class SimpleNixClient:
"""Simplified client for interacting with NIX via cleos and JSON"""
def __init__(self,
rodeos_api: str = None,
nodeos_api: str = None,
environment: str = None,
timeout: int = 30):
"""
Initialize NIX client
Args:
rodeos_api: Rodeos API endpoint (overrides environment)
nodeos_api: Nodeos API endpoint (overrides environment)
environment: Environment name (dev, uat, prod, etc.)
timeout: Request timeout in seconds
"""
# If explicit endpoints are provided, use them
if rodeos_api and nodeos_api:
self.rodeos_api = rodeos_api
self.nodeos_api = nodeos_api
# Otherwise, use environment configuration
else:
env_config = EnvironmentConfig()
# Use provided environment, or check NODEOS_ENV, or default to 'cdev'
env = environment or os.getenv("NODEOS_ENV", "cdev")
nodeos_endpoint, rodeos_endpoint = env_config.get_endpoints(env)
# Use environment-specific endpoints, don't check os.getenv() here
# because it might have localhost values
self.rodeos_api = rodeos_api or rodeos_endpoint
self.nodeos_api = nodeos_api or nodeos_endpoint
logger.debug(f"Using environment '{env}': nodeos={self.nodeos_api}, rodeos={self.rodeos_api}")
self.timeout = timeout
# Try to find cleos
self.cleos_cmd = self._find_cleos()
def _find_cleos(self) -> str:
"""Find cleos binary in common locations"""
# Check if cleos is in PATH
result = subprocess.run(["which", "cleos"], capture_output=True, text=True)
if result.returncode == 0:
return result.stdout.strip()
# Check common locations
common_paths = [
"/usr/local/bin/cleos",
"/usr/bin/cleos",
os.path.expanduser("~/workspace/taurus-node/build/bin/cleos"),
os.getenv("CLEOS_PATH", "")
]
for path in common_paths:
if path and os.path.exists(path):
return path
# Default to cleos and hope it's in PATH
return "cleos"
async def query(self,
contract: str,
action: str,
params: Union[Dict[str, Any], str]) -> Any:
"""
Execute a query action with JSON parameters
Args:
contract: Contract name (e.g., 'nix.q')
action: Action name
params: JSON parameters dict or path to JSON file
Returns:
Query result (JSON response from contract)
"""
# Handle params - can be dict, JSON string, or file path
if isinstance(params, str):
# Check if it's a file path
if params.endswith('.json') and os.path.exists(params):
logger.debug(f"Reading params from file: {params}")
with open(params, 'r') as f:
json_params = f.read()
else:
# Assume it's already a JSON string
json_params = params
elif params:
# Convert dict to JSON string
json_params = json.dumps(params)
else:
json_params = "{}"
# Build cleos command
cmd = [
self.cleos_cmd, "-u", self.rodeos_api,
"push", "action",
"--use-old-send-rpc",
"--return-failure-trace",
"0", # account (0 for read-only)
contract,
action,
json_params,
"-s" # silent with JSON output
]
try:
logger.debug(f"Executing: {' '.join(cmd)}")
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=self.timeout,
check=False
)
if result.returncode != 0:
error_msg = result.stderr or result.stdout
logger.error(f"cleos error: {error_msg}")
# Try to parse error as JSON for better error messages
try:
error_json = json.loads(error_msg)
if "error" in error_json:
raise Exception(f"Query failed: {error_json['error']}")
except json.JSONDecodeError:
pass
raise Exception(f"cleos failed: {error_msg}")
# Try to parse as JSON first, but return raw stdout if it fails
try:
response = json.loads(result.stdout)
return response
except json.JSONDecodeError:
# Return raw stdout if not valid JSON
logger.debug(f"Response is not JSON, returning raw output")
return result.stdout
except subprocess.TimeoutExpired:
logger.error(f"cleos timeout calling {action}")
raise Exception(f"Query timeout after {self.timeout} seconds")
except Exception as e:
logger.error(f"Error calling {action}: {e}")
raise
def generate_json_example(self, query_name: str, contract: str = "nix.q") -> Dict[str, Any]:
"""
Generate a JSON example for a query using ABI resolver
Args:
query_name: Name of the query
contract: Contract name
Returns:
Example JSON structure from ABI
"""
try:
# Try to use cached ABI first
from pathlib import Path
cache_file = Path(f".abi_cache/{contract}.json")
if cache_file.exists():
with open(cache_file, 'r') as f:
abi = json.load(f)
from .abi_resolver import ABIResolver
resolver = ABIResolver(abi_data=abi)
action_info = resolver.resolve_action(query_name)
return action_info.get("example", {})
else:
# Fall back to fetching ABI
from .abi_fetcher import ABIFetcher
fetcher = ABIFetcher(nodeos_api=self.nodeos_api)
return fetcher.get_action_template(contract, query_name)
except Exception as e:
logger.debug(f"Could not generate example from ABI: {e}")
# Return empty object as fallback
return {}