Complete implementation of asset management CLI commands with comprehensive user experience improvements: ## Core Features - Asset management commands: add, list, stats, cleanup - Package management commands: create, extract, list, validate - Workspace management commands: init, status, sync ## CLI Integration - Seamless integration with existing markitect CLI patterns - Consistent Click command group registration - Professional output formatting with checkmarks and structured details - Comprehensive help text with examples and feature descriptions ## Code Quality - Extracted common CLI utilities for consistent UX patterns - Robust error handling with informative messages - Configuration integration with sensible defaults - Path validation and workspace management ## Testing & Quality Assurance - Comprehensive integration tests covering all command groups - No regressions in existing CLI functionality - End-to-end workflow validation - Production-ready error handling and edge cases ## Documentation - Enhanced docstrings with usage examples - Comprehensive --help text for all commands - Clear argument descriptions and feature highlights 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
336 lines
9.7 KiB
Python
336 lines
9.7 KiB
Python
"""
|
|
CLI utilities for MarkiTect command-line interface.
|
|
|
|
This module provides common utilities and patterns used across CLI commands:
|
|
- Output formatting (table, JSON)
|
|
- Error handling decorators
|
|
- Common Click options
|
|
- Configuration loading helpers
|
|
|
|
Used by asset management commands and can be extended for other CLI modules.
|
|
"""
|
|
|
|
import click
|
|
import json
|
|
import sys
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
from tabulate import tabulate
|
|
from typing import Any, Dict, List, Optional, Callable
|
|
|
|
# Import for configuration support
|
|
try:
|
|
from .config_manager import ConfigurationManager
|
|
CONFIG_AVAILABLE = True
|
|
except ImportError:
|
|
CONFIG_AVAILABLE = False
|
|
|
|
|
|
def format_table_output(data: List[Dict[str, Any]], headers: List[str],
|
|
tablefmt: str = 'grid') -> str:
|
|
"""Format data as table for console output.
|
|
|
|
Args:
|
|
data: List of dictionaries containing row data
|
|
headers: List of column headers
|
|
tablefmt: Table format style (default: 'grid')
|
|
|
|
Returns:
|
|
Formatted table string
|
|
"""
|
|
if not data:
|
|
return "No data to display"
|
|
|
|
# Convert dict data to list of lists for tabulate
|
|
table_data = []
|
|
for item in data:
|
|
row = [item.get(header.lower(), item.get(header, 'N/A')) for header in headers]
|
|
table_data.append(row)
|
|
|
|
return tabulate(table_data, headers=headers, tablefmt=tablefmt)
|
|
|
|
|
|
def format_json_output(data: Any, indent: int = 2) -> str:
|
|
"""Format data as JSON for programmatic consumption.
|
|
|
|
Args:
|
|
data: Data to format as JSON
|
|
indent: JSON indentation level
|
|
|
|
Returns:
|
|
JSON formatted string
|
|
"""
|
|
return json.dumps(data, indent=indent, default=str)
|
|
|
|
|
|
def handle_asset_errors(func: Callable) -> Callable:
|
|
"""Decorator to handle common asset management errors.
|
|
|
|
Provides consistent error handling for asset-related CLI commands.
|
|
"""
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except ImportError as e:
|
|
if "assets" in str(e).lower():
|
|
click.echo("Error: Asset management backend not available", err=True)
|
|
click.echo("Ensure markitect.assets module is properly installed", err=True)
|
|
else:
|
|
click.echo(f"Import error: {e}", err=True)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
# Import asset exceptions if available
|
|
try:
|
|
from .assets import AssetError, PackagingError
|
|
if isinstance(e, (AssetError, PackagingError)):
|
|
click.echo(f"Asset error: {e}", err=True)
|
|
else:
|
|
click.echo(f"Unexpected error: {e}", err=True)
|
|
except ImportError:
|
|
click.echo(f"Unexpected error: {e}", err=True)
|
|
sys.exit(1)
|
|
|
|
return wrapper
|
|
|
|
|
|
def require_workspace(func: Callable) -> Callable:
|
|
"""Decorator to ensure workspace exists before running command.
|
|
|
|
Checks for workspace directory and shows helpful message if not found.
|
|
"""
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs):
|
|
workspace_dir = Path.cwd() / "markitect_workspace"
|
|
if not workspace_dir.exists():
|
|
click.echo("No workspace found in current directory", err=True)
|
|
click.echo("Run 'markitect workspace init' to create one", err=True)
|
|
sys.exit(1)
|
|
return func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
# Common Click options
|
|
def output_format_option(default: str = 'table'):
|
|
"""Common output format option for list commands."""
|
|
return click.option(
|
|
'--format', 'output_format',
|
|
type=click.Choice(['table', 'json']),
|
|
default=default,
|
|
help=f'Output format (default: {default})'
|
|
)
|
|
|
|
|
|
def dry_run_option():
|
|
"""Common dry-run option for potentially destructive commands."""
|
|
return click.option(
|
|
'--dry-run', is_flag=True,
|
|
help='Show what would be done without making changes'
|
|
)
|
|
|
|
|
|
def verbose_option():
|
|
"""Common verbose option for detailed output."""
|
|
return click.option(
|
|
'--verbose', '-v', is_flag=True,
|
|
help='Enable verbose output'
|
|
)
|
|
|
|
|
|
class ClickOutputFormatter:
|
|
"""
|
|
Helper class for consistent CLI output formatting across MarkiTect commands.
|
|
|
|
Provides standardized methods for displaying success, info, warning, and error
|
|
messages with consistent formatting including icons and structured details.
|
|
|
|
Usage:
|
|
ClickOutputFormatter.success("Operation completed", {"Files": 5})
|
|
ClickOutputFormatter.error("Failed to process")
|
|
"""
|
|
|
|
@staticmethod
|
|
def success(message: str, details: Optional[Dict[str, Any]] = None):
|
|
"""
|
|
Display success message with checkmark and optional details.
|
|
|
|
Args:
|
|
message: Success message to display
|
|
details: Optional dictionary of key-value details to show
|
|
"""
|
|
click.echo(f"✓ {message}")
|
|
if details:
|
|
for key, value in details.items():
|
|
click.echo(f" {key}: {value}")
|
|
|
|
@staticmethod
|
|
def info(message: str, details: Optional[Dict[str, Any]] = None):
|
|
"""
|
|
Display informational message with optional details.
|
|
|
|
Args:
|
|
message: Info message to display
|
|
details: Optional dictionary of key-value details to show
|
|
"""
|
|
click.echo(message)
|
|
if details:
|
|
for key, value in details.items():
|
|
click.echo(f" {key}: {value}")
|
|
|
|
@staticmethod
|
|
def warning(message: str):
|
|
"""
|
|
Display warning message with warning icon.
|
|
|
|
Args:
|
|
message: Warning message to display
|
|
"""
|
|
click.echo(f"⚠ {message}", err=True)
|
|
|
|
@staticmethod
|
|
def error(message: str, exit_code: int = 1):
|
|
"""
|
|
Display error message with error icon and exit.
|
|
|
|
Args:
|
|
message: Error message to display
|
|
exit_code: Exit code to use (default: 1)
|
|
"""
|
|
click.echo(f"✗ {message}", err=True)
|
|
sys.exit(exit_code)
|
|
|
|
@staticmethod
|
|
def table(data: List[Dict[str, Any]], headers: List[str]):
|
|
"""Display data as formatted table."""
|
|
if not data:
|
|
click.echo("No data to display")
|
|
return
|
|
|
|
table_output = format_table_output(data, headers)
|
|
click.echo(table_output)
|
|
|
|
@staticmethod
|
|
def json_output(data: Any):
|
|
"""Display data as JSON."""
|
|
json_output = format_json_output(data)
|
|
click.echo(json_output)
|
|
|
|
|
|
def get_configuration() -> Optional[Dict[str, Any]]:
|
|
"""Get current markitect configuration.
|
|
|
|
Returns:
|
|
Configuration dictionary if available, None otherwise
|
|
"""
|
|
if not CONFIG_AVAILABLE:
|
|
return None
|
|
|
|
try:
|
|
config_manager = ConfigurationManager()
|
|
return config_manager.get_config()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_asset_config() -> Dict[str, Any]:
|
|
"""Get asset management configuration with defaults.
|
|
|
|
Returns:
|
|
Asset configuration dictionary with sensible defaults
|
|
"""
|
|
config = get_configuration()
|
|
|
|
if config and 'asset_management' in config:
|
|
asset_config = config['asset_management']
|
|
else:
|
|
asset_config = {}
|
|
|
|
# Apply defaults
|
|
defaults = {
|
|
'enabled': True,
|
|
'workspace_path': './markitect_workspace',
|
|
'shared_assets_path': './markitect_workspace/shared_assets',
|
|
'packages_path': './markitect_workspace/packages',
|
|
'auto_dedupe': True,
|
|
'symlink_preferred': True,
|
|
'fallback_to_copy': True,
|
|
'compression_level': 6,
|
|
'include_manifest': True,
|
|
'validate_on_create': True,
|
|
'cache_enabled': True,
|
|
'batch_size': 100,
|
|
'max_file_size_mb': 50
|
|
}
|
|
|
|
# Merge with defaults
|
|
for key, default_value in defaults.items():
|
|
if key not in asset_config:
|
|
asset_config[key] = default_value
|
|
|
|
return asset_config
|
|
|
|
|
|
def validate_file_path(path: str, must_exist: bool = True) -> Path:
|
|
"""Validate and normalize file path.
|
|
|
|
Args:
|
|
path: File path string
|
|
must_exist: Whether file must exist
|
|
|
|
Returns:
|
|
Validated Path object
|
|
|
|
Raises:
|
|
click.ClickException: If validation fails
|
|
"""
|
|
file_path = Path(path).resolve()
|
|
|
|
if must_exist and not file_path.exists():
|
|
raise click.ClickException(f"File not found: {file_path}")
|
|
|
|
if must_exist and file_path.is_dir():
|
|
raise click.ClickException(f"Expected file, got directory: {file_path}")
|
|
|
|
return file_path
|
|
|
|
|
|
def validate_directory_path(path: str, must_exist: bool = True,
|
|
create_if_missing: bool = False) -> Path:
|
|
"""Validate and normalize directory path.
|
|
|
|
Args:
|
|
path: Directory path string
|
|
must_exist: Whether directory must exist
|
|
create_if_missing: Whether to create directory if missing
|
|
|
|
Returns:
|
|
Validated Path object
|
|
|
|
Raises:
|
|
click.ClickException: If validation fails
|
|
"""
|
|
dir_path = Path(path).resolve()
|
|
|
|
if not dir_path.exists():
|
|
if create_if_missing:
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
elif must_exist:
|
|
raise click.ClickException(f"Directory not found: {dir_path}")
|
|
elif dir_path.exists() and not dir_path.is_dir():
|
|
raise click.ClickException(f"Expected directory, got file: {dir_path}")
|
|
|
|
return dir_path
|
|
|
|
|
|
def confirm_destructive_action(message: str, default: bool = False) -> bool:
|
|
"""Prompt user to confirm destructive action.
|
|
|
|
Args:
|
|
message: Confirmation message
|
|
default: Default choice if user just presses enter
|
|
|
|
Returns:
|
|
True if user confirms, False otherwise
|
|
"""
|
|
return click.confirm(message, default=default) |