fix: resolve CLI config package shadowing config.py module
Some checks failed
CLI Tests / test-cli (push) Failing after 7s
Deploy to Testnet / deploy-testnet (push) Successful in 1m17s
Security Scanning / security-scan (push) Successful in 38s
Multi-Node Stress Testing / stress-test (push) Successful in 2s
Cross-Node Transaction Testing / transaction-test (push) Successful in 2s

Renamed cli/config/ directory to cli/config_data/ to prevent package
shadowing of cli/aitbc_cli/config.py module. Removed problematic
sys.modules manipulation in aitbc_cli/__init__.py that was causing
circular import issues. Now 'from aitbc_cli.config import CLIConfig' works correctly.
This commit is contained in:
aitbc
2026-05-19 19:53:34 +02:00
parent 55594ca9ee
commit 26f7d72b8e
3 changed files with 241 additions and 2 deletions

View File

@@ -14,8 +14,7 @@ if "aitbc_cli.core" not in sys.modules:
sys.modules["aitbc_cli.core"] = import_module("core") sys.modules["aitbc_cli.core"] = import_module("core")
if "aitbc_cli.models" not in sys.modules: if "aitbc_cli.models" not in sys.modules:
sys.modules["aitbc_cli.models"] = import_module("models") sys.modules["aitbc_cli.models"] = import_module("models")
if "aitbc_cli.config" not in sys.modules: # Note: aitbc_cli.config is imported normally to avoid circular import issues
sys.modules["aitbc_cli.config"] = import_module("config")
__all__ = ["cli", "main"] __all__ = ["cli", "main"]

114
cli/config_data/__init__.py Executable file
View File

@@ -0,0 +1,114 @@
"""Configuration management for AITBC CLI"""
import os
import yaml
from pathlib import Path
from typing import Optional
from dataclasses import dataclass, field
from dotenv import load_dotenv
@dataclass
class Config:
"""Configuration object for AITBC CLI"""
coordinator_url: str = "http://127.0.0.1:8011"
api_key: Optional[str] = None
role: Optional[str] = None # admin, client, miner, etc.
config_dir: Path = field(default_factory=lambda: Path.home() / ".aitbc")
config_file: Optional[str] = None
blockchain_rpc_url: str = "http://127.0.0.1:8006"
wallet_url: str = "http://127.0.0.1:8002"
def _validate_localhost_urls(self):
"""Validate that all service URLs point to localhost"""
localhost_prefixes = ["http://localhost:", "http://127.0.0.1:", "https://localhost:", "https://127.0.0.1:"]
urls_to_check = [
("coordinator_url", self.coordinator_url),
("blockchain_rpc_url", self.blockchain_rpc_url),
("wallet_url", self.wallet_url)
]
for url_name, url in urls_to_check:
if not any(url.startswith(prefix) for prefix in localhost_prefixes):
# Force to localhost if not already
if url_name == "coordinator_url":
self.coordinator_url = "http://localhost:8011"
elif url_name == "blockchain_rpc_url":
self.blockchain_rpc_url = "http://localhost:8006"
elif url_name == "wallet_url":
self.wallet_url = "http://localhost:8002"
def __post_init__(self):
"""Initialize configuration"""
# Load environment variables
load_dotenv()
# Set default config file based on role if not specified
if not self.config_file:
if self.role:
self.config_file = str(self.config_dir / f"{self.role}-config.yaml")
else:
self.config_file = str(self.config_dir / "config.yaml")
# Load config from file if it exists
self.load_from_file()
# Override with environment variables
if os.getenv("AITBC_URL"):
self.coordinator_url = os.getenv("AITBC_URL")
if os.getenv("AITBC_API_KEY"):
self.api_key = os.getenv("AITBC_API_KEY")
if os.getenv("AITBC_ROLE"):
self.role = os.getenv("AITBC_ROLE")
if os.getenv("AITBC_BLOCKCHAIN_RPC_URL"):
self.blockchain_rpc_url = os.getenv("AITBC_BLOCKCHAIN_RPC_URL")
if os.getenv("AITBC_WALLET_URL"):
self.wallet_url = os.getenv("AITBC_WALLET_URL")
# Validate and enforce localhost URLs
self._validate_localhost_urls()
def load_from_file(self):
"""Load configuration from YAML file"""
if self.config_file and Path(self.config_file).exists():
try:
with open(self.config_file, 'r') as f:
data = yaml.safe_load(f) or {}
self.coordinator_url = data.get('coordinator_url', self.coordinator_url)
self.api_key = data.get('api_key', self.api_key)
self.role = data.get('role', self.role)
self.blockchain_rpc_url = data.get('blockchain_rpc_url', self.blockchain_rpc_url)
self.wallet_url = data.get('wallet_url', self.wallet_url)
except Exception as e:
print(f"Warning: Could not load config file: {e}")
# Validate and enforce localhost URLs after file loading
self._validate_localhost_urls()
def save_to_file(self):
"""Save configuration to YAML file"""
if not self.config_file:
return
# Ensure config directory exists
Path(self.config_file).parent.mkdir(parents=True, exist_ok=True)
data = {
'coordinator_url': self.coordinator_url,
'api_key': self.api_key,
'blockchain_rpc_url': self.blockchain_rpc_url,
'wallet_url': self.wallet_url
}
if self.role:
data['role'] = self.role
with open(self.config_file, 'w') as f:
yaml.dump(data, f, default_flow_style=False)
def get_config(config_file: Optional[str] = None, role: Optional[str] = None) -> Config:
"""Get configuration instance with optional role"""
return Config(config_file=config_file, role=role)

126
cli/config_data/chains.py Normal file
View File

@@ -0,0 +1,126 @@
"""
Chain registry configuration for AITBC CLI
Manages available blockchain networks and their configurations
"""
from typing import Dict, Optional
from dataclasses import dataclass
import os
@dataclass
class ChainConfig:
"""Configuration for a blockchain network"""
chain_id: str
name: str
rpc_url: str
explorer_url: Optional[str] = None
is_testnet: bool = False
native_currency: str = "AITBC"
class ChainRegistry:
"""Registry for managing blockchain network configurations"""
def __init__(self):
"""Initialize chain registry with default chains"""
self.chains: Dict[str, ChainConfig] = {}
self._load_default_chains()
def _load_default_chains(self) -> None:
"""Load default chain configurations"""
# AITBC Devnet
self.chains["ait-devnet"] = ChainConfig(
chain_id="ait-devnet",
name="AITBC Development Network",
rpc_url="http://localhost:8025",
explorer_url="http://localhost:8026",
is_testnet=True,
native_currency="AITBC"
)
# AITBC Testnet
self.chains["ait-testnet"] = ChainConfig(
chain_id="ait-testnet",
name="AITBC Test Network",
rpc_url="http://localhost:8027",
explorer_url="http://localhost:8028",
is_testnet=True,
native_currency="AITBC"
)
def get_chain(self, chain_id: str) -> Optional[ChainConfig]:
"""Get chain configuration by ID"""
return self.chains.get(chain_id)
def get_all_chains(self) -> Dict[str, ChainConfig]:
"""Get all registered chains"""
return self.chains.copy()
def get_chain_ids(self) -> list[str]:
"""Get list of all chain IDs"""
return list(self.chains.keys())
def get_testnet_chains(self) -> Dict[str, ChainConfig]:
"""Get all testnet chains"""
return {
chain_id: config
for chain_id, config in self.chains.items()
if config.is_testnet
}
def get_mainnet_chains(self) -> Dict[str, ChainConfig]:
"""Get all mainnet chains"""
return {
chain_id: config
for chain_id, config in self.chains.items()
if not config.is_testnet
}
def register_chain(self, chain_id: str, config: ChainConfig) -> None:
"""Register a new chain configuration"""
self.chains[chain_id] = config
def unregister_chain(self, chain_id: str) -> bool:
"""Unregister a chain configuration"""
if chain_id in self.chains:
del self.chains[chain_id]
return True
return False
def load_from_env(self) -> None:
"""Load additional chains from environment variables"""
# Format: AITBC_CHAIN_<CHAIN_ID>_NAME, AITBC_CHAIN_<CHAIN_ID>_RPC_URL, etc.
for key, value in os.environ.items():
if key.startswith("AITBC_CHAIN_") and key.endswith("_RPC_URL"):
chain_id = key.replace("AITBC_CHAIN_", "").replace("_RPC_URL", "").lower()
name = os.environ.get(f"AITBC_CHAIN_{chain_id.upper()}_NAME", chain_id)
explorer_url = os.environ.get(f"AITBC_CHAIN_{chain_id.upper()}_EXPLORER_URL")
is_testnet = os.environ.get(f"AITBC_CHAIN_{chain_id.upper()}_IS_TESTNET", "false").lower() == "true"
native_currency = os.environ.get(f"AITBC_CHAIN_{chain_id.upper()}_NATIVE_CURRENCY", "AITBC")
self.register_chain(
chain_id,
ChainConfig(
chain_id=chain_id,
name=name,
rpc_url=value,
explorer_url=explorer_url,
is_testnet=is_testnet,
native_currency=native_currency
)
)
# Global chain registry instance
_chain_registry: Optional[ChainRegistry] = None
def get_chain_registry() -> ChainRegistry:
"""Get or create global chain registry instance"""
global _chain_registry
if _chain_registry is None:
_chain_registry = ChainRegistry()
_chain_registry.load_from_env()
return _chain_registry