fix: code cleanup

Author: yanbasic

src/emd/commands/bootstrap.py

@@ -25,7 +25,7 @@ def bootstrap(
 ):
     """
     Manually initialize AWS resources for model deployment.
-    
+
     Note: This command is optional. Running 'emd deploy' will automatically
     set up infrastructure when needed.
     """

src/emd/utils/decorators.py

@@ -113,11 +113,11 @@ def print_aws_config():
     response = sts.get_caller_identity()
     account_id = response["Account"]
     region = get_current_region()
-    
+
     if region is None:
         console.print("[yellow]warning: Unable to determine AWS region.[/yellow]")
         raise typer.Exit(1)
-    
+
     console.print(Panel.fit(
         f"[bold green]Account ID:[/bold green] {account_id}\n"
         f"[bold green]Region    :[/bold green] {region}",
@@ -136,12 +136,12 @@ def wrapper(*args, **kwargs):
             error_msg = str(e)
             logger.error(f"AWS configuration loading failed: {error_msg}")
             logger.debug(traceback.format_exc())
-            
+
             if kwargs.get("allow_local_deploy") or kwargs.get("only_allow_local_deploy", False):
                 logger.warning(f"AWS configuration error: {error_msg}. Proceeding with local deployment.")
                 kwargs['only_allow_local_deploy'] = True
                 return fn(*args, **kwargs)
-            
+
             console = Console()
             console.print(f"[red]❌ AWS Configuration Error: {error_msg}[/red]")
             raise typer.Exit(1)

src/emd/utils/smart_bootstrap.py

@@ -18,7 +18,7 @@ def get_deployed_infrastructure_version(region: str) -> Optional[str]:
     try:
         cfn = boto3.client('cloudformation', region_name=region)
         stack_info = cfn.describe_stacks(StackName=ENV_STACK_NAME)['Stacks'][0]
-        
+
         # Get ArtifactVersion parameter from stack
         for param in stack_info.get('Parameters', []):
             if param['ParameterKey'] == 'ArtifactVersion':
@@ -36,20 +36,20 @@ def check_infrastructure_completeness(region: str) -> Tuple[bool, str]:
     try:
         cfn = boto3.client('cloudformation', region_name=region)
         s3 = boto3.client('s3', region_name=region)
-        
+
         # Check CloudFormation stack
         try:
             stack_info = cfn.describe_stacks(StackName=ENV_STACK_NAME)['Stacks'][0]
             stack_status = stack_info['StackStatus']
-            
+
             if stack_status not in ['CREATE_COMPLETE', 'UPDATE_COMPLETE']:
                 return False, f"CloudFormation stack status: {stack_status}"
-                
+
         except cfn.exceptions.ClientError as e:
             if "does not exist" in str(e):
                 return False, "CloudFormation stack does not exist"
             raise
-        
+
         # Check S3 bucket (get bucket name from stack resources)
         try:
             from emd.sdk.bootstrap import get_bucket_name
@@ -60,9 +60,9 @@ def check_infrastructure_completeness(region: str) -> Tuple[bool, str]:
             s3.head_bucket(Bucket=bucket_name)
         except Exception as e:
             return False, f"S3 bucket issue: {str(e)}"
-        
+
         return True, "Infrastructure is complete"
-        
+
     except Exception as e:
         logger.debug(f"Infrastructure completeness check failed: {e}")
         return False, f"Infrastructure check failed: {str(e)}"
@@ -71,7 +71,7 @@ def check_infrastructure_completeness(region: str) -> Tuple[bool, str]:
 class SmartBootstrapManager:
     def __init__(self):
         self.console = Console()
-        
+
     def check_version_compatibility(self, current_version: str, deployed_version: str, region: str) -> str:
         """
         Check version compatibility and infrastructure completeness
@@ -82,14 +82,14 @@ def check_version_compatibility(self, current_version: str, deployed_version: st
         if not is_complete:
             logger.debug(f"Infrastructure incomplete: {status_msg}")
             return 'auto_bootstrap'  # Infrastructure missing/incomplete, need bootstrap
-        
+
         if not deployed_version:
             return 'auto_bootstrap'  # No version info, need bootstrap
-        
+
         try:
             current_parsed = packaging.version.parse(current_version)
             deployed_parsed = packaging.version.parse(deployed_version)
-            
+
             if current_parsed > deployed_parsed:
                 return 'auto_bootstrap'  # Local newer, auto bootstrap
             elif deployed_parsed > current_parsed:
@@ -99,7 +99,7 @@ def check_version_compatibility(self, current_version: str, deployed_version: st
         except Exception as e:
             logger.debug(f"Failed to parse versions: {e}")
             return 'auto_bootstrap'  # Default to bootstrap if parsing fails
-    
+
     def show_bootstrap_notification(self, current_version: str, deployed_version: str):
         """Show notification about automatic bootstrap"""
         self.console.print()  # Empty line for spacing
@@ -108,18 +108,18 @@ def show_bootstrap_notification(self, current_version: str, deployed_version: st
         else:
             self.console.print(f"🚀 [bold green]Setting up infrastructure...[/bold green] [bold green]{current_version}[/bold green]")
         self.console.print()  # Empty line for spacing
-    
+
     def show_version_mismatch_warning(self, current_version: str, deployed_version: str):
         """Show warning when cloud version is newer than local version"""
         self.console.print()  # Empty line for spacing
         self.console.print(f"⚠️  [bold yellow]Version mismatch:[/bold yellow] Local [dim]{current_version}[/dim] < Cloud [bold yellow]{deployed_version}[/bold yellow]")
         self.console.print(f"   [bold]Recommendation:[/bold] pip install --upgrade easy-model-deployer")
         self.console.print()  # Empty line for spacing
-    
+
     def is_auto_bootstrap_disabled(self) -> bool:
         """Check if auto bootstrap is disabled via environment variable"""
         return os.getenv("EMD_DISABLE_AUTO_BOOTSTRAP", "").lower() in ["true", "1", "yes"]
-    
+
     def auto_bootstrap_if_needed(self, region: str) -> bool:
         """
         Automatically run bootstrap if needed based on comprehensive infrastructure check
@@ -129,67 +129,67 @@ def auto_bootstrap_if_needed(self, region: str) -> bool:
         if self.is_auto_bootstrap_disabled():
             logger.debug("Auto bootstrap disabled via EMD_DISABLE_AUTO_BOOTSTRAP")
             return False
-        
+
         current_version = VERSION
         deployed_version = get_deployed_infrastructure_version(region)
-        
+
         action = self.check_version_compatibility(current_version, deployed_version, region)
-        
+
         if action == 'compatible':
             return False  # No action needed
-        
+
         elif action == 'version_mismatch_warning':
             # Cloud version > Local version - show warning and ask user
             self.show_version_mismatch_warning(current_version, deployed_version)
-            
+
             if not typer.confirm("Continue deployment despite version mismatch?", default=False):
                 self.console.print("[yellow]Deployment cancelled. Please update EMD to the latest version.[/yellow]")
                 raise typer.Exit(0)
-            
+
             return False  # User chose to continue, no bootstrap
-        
+
         elif action == 'auto_bootstrap':
             # Infrastructure missing/incomplete OR version mismatch - ask for confirmation
             self.show_bootstrap_notification(current_version, deployed_version)
-            
+
             # Ask for user confirmation
             if deployed_version:
                 # Update scenario
                 confirm_msg = f"Update infrastructure from {deployed_version} to {current_version}?"
             else:
                 # Initialize scenario
                 confirm_msg = f"Initialize EMD infrastructure for version {current_version}?"
-            
+
             if not typer.confirm(confirm_msg, default=True):
                 self.console.print("[yellow]Bootstrap cancelled. Infrastructure will not be updated.[/yellow]")
                 self.console.print("[red]Deployment cannot proceed without compatible infrastructure.[/red]")
                 raise typer.Exit(1)
-            
+
             # User confirmed - proceed with bootstrap
             try:
                 from emd.sdk.bootstrap import create_env_stack, get_bucket_name
-                
+
                 bucket_name = get_bucket_name(
                     bucket_prefix=ENV_BUCKET_NAME_PREFIX,
                     region=region
                 )
-                
+
                 create_env_stack(
                     region=region,
                     stack_name=ENV_STACK_NAME,
                     bucket_name=bucket_name,
                     force_update=True
                 )
-                
+
                 self.console.print("[bold green]✅ Infrastructure setup completed successfully![/bold green]")
                 return True
-                
+
             except Exception as e:
                 self.console.print(f"[bold red]❌ Infrastructure setup failed: {str(e)}[/bold red]")
                 raise typer.Exit(1)
-        
+
         return False
 
 
 # Global instance
-smart_bootstrap_manager = SmartBootstrapManager()
+smart_bootstrap_manager = SmartBootstrapManager()

src/emd/utils/update_notifier.py

@@ -18,23 +18,23 @@ def __init__(self):
         self.console = Console()
         self.cache_file = Path.home() / ".emd_update_cache.json"
         self.cache_duration = 3600  # 1 hour cache
-        
+
     def is_development_mode(self) -> bool:
         """Check if running in development mode"""
         return VERSION == "0.0.0"
-    
+
     def should_check_updates(self) -> bool:
         """Determine if we should check for updates"""
         # Skip in development mode
         if self.is_development_mode():
             return False
-        
+
         # Skip if user disabled checks
         if os.getenv("EMD_DISABLE_UPDATE_CHECK", "").lower() in ["true", "1", "yes"]:
             return False
-            
+
         return True
-    
+
     def get_cached_version_info(self) -> Optional[Dict[str, Any]]:
         """Get cached version information"""
         try:
@@ -46,7 +46,7 @@ def get_cached_version_info(self) -> Optional[Dict[str, Any]]:
         except Exception as e:
             logger.debug(f"Failed to read update cache: {e}")
         return None
-    
+
     def cache_version_info(self, latest_version: str, has_update: bool):
         """Cache version information"""
         try:
@@ -60,30 +60,30 @@ def cache_version_info(self, latest_version: str, has_update: bool):
                 json.dump(data, f)
         except Exception as e:
             logger.debug(f"Failed to write update cache: {e}")
-    
+
     def fetch_latest_version(self) -> Optional[str]:
         """Fetch latest version from PyPI"""
         try:
             response = requests.get(
-                "https://pypi.org/pypi/easy-model-deployer/json", 
+                "https://pypi.org/pypi/easy-model-deployer/json",
                 timeout=2  # Quick timeout to avoid blocking
             )
             if response.status_code == 200:
                 return response.json()["info"]["version"]
         except Exception as e:
             logger.debug(f"Failed to fetch latest version: {e}")
         return None
-    
+
     def check_for_updates(self) -> Tuple[bool, Optional[str]]:
         """Check if updates are available with caching"""
         if not self.should_check_updates():
             return False, None
-        
+
         # Try cache first
         cached = self.get_cached_version_info()
         if cached and cached.get('current_version') == VERSION:
             return cached['has_update'], cached.get('latest_version')
-        
+
         # Fetch from PyPI
         latest = self.fetch_latest_version()
         if latest:
@@ -93,14 +93,14 @@ def check_for_updates(self) -> Tuple[bool, Optional[str]]:
                 return has_update, latest
             except Exception as e:
                 logger.debug(f"Failed to parse versions: {e}")
-        
+
         return False, None
-    
+
     def show_update_notification(self):
         """Display update notification if available"""
         try:
             has_update, latest_version = self.check_for_updates()
-            
+
             if has_update and latest_version:
                 # Display simple colored message
                 self.console.print()  # Empty line for spacing
@@ -113,4 +113,4 @@ def show_update_notification(self):
 
 
 # Global instance
-update_notifier = UpdateNotifier()
+update_notifier = UpdateNotifier()