Enhance logging and error handling in configuration and command execution

Author: Domains18

cloudsql_to_supabase/config.py

@@ -1,38 +1,50 @@
 from dotenv import load_dotenv
 import os
+from pathlib import Path
+import logging 
 
+logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s')
+
+
+logger = logging.getLogger('cloudsql_to_supabase')
 
 load_dotenv()
 
+
 CLOUDSQL_USER = os.getenv("CLOUDSQL_USER")
 CLOUDSQL_HOST = os.getenv("CLOUDSQL_HOST")
 CLOUDSQL_DB = os.getenv("CLOUDSQL_DB")
-SUPABASE_USER = os.getenv("SUPABASE_USER")
+CLOUDSQL_PORT = int(os.getenv("CLOUDSQL_PORT", 5432))
+CLOUDSQL_SSL_MODE = os.getenv("CLOUDSQL_SSL_MODE", "prefer")
+
+SUPABASE_USER = os.getenv("SUPABASE_USER", "postgres")
 SUPABASE_HOST = os.getenv("SUPABASE_HOST")
 SUPABASE_DB = os.getenv("SUPABASE_DB", "postgres")
 SUPABASE_PASSWORD = os.getenv("SUPABASE_PASSWORD")
 SUPABASE_PORT = int(os.getenv("SUPABASE_PORT", 5432))
-OUTPUT_DUMP = "backup.sql"
-CLEANED_DUMP = "cleaned_backup.sql"
+SUPABASE_SSL_MODE = os.getenv("SUPABASE_SSL_MODE", "require")
+
+OUTPUT_DIR = Path(os.getenv('OUTPUT_DIR', '.'))
+OUTPUT_DUMP = OUTPUT_DIR / os.getenv("OUTPUT_DUMP", "backup.sql")
+CLEANED_DUMP = OUTPUT_DIR / os.getenv("CLEANED_DUMP", "cleaned_backup.sql")
 
 
 def validate_config():
     """
     Validate the configuration values.
     """
-    required_vars = [
-        "CLOUDSQL_USER",
-        "CLOUDSQL_HOST",
-        "CLOUDSQL_DB",
-        "SUPABASE_USER",
-        "SUPABASE_HOST",
-        "SUPABASE_DB",
-        "SUPABASE_PASSWORD",
-        "SUPABASE_PORT",
-    ]
-    for var in required_vars:
-        if not os.getenv(var):
-            raise ValueError(f"Environment variable {var} is not set.")
-    if not SUPABASE_PORT or SUPABASE_PORT <= 0:
-        raise ValueError("SUPABASE_PORT must be a positive integer.")
-    
+    required_vars = {
+         "CLOUDSQL_USER": CLOUDSQL_USER,
+        "CLOUDSQL_HOST": CLOUDSQL_HOST,
+        "CLOUDSQL_DB": CLOUDSQL_DB,
+        "SUPABASE_HOST": SUPABASE_HOST,
+        "SUPABASE_PASSWORD": SUPABASE_PASSWORD,
+    }
+    
+    
+    missing = [var for var, value in required_vars.items() if not value]
+    
+    if missing:
+        raise ValueError(f"Missing required environment variables: {', '.join(missing)}")
+    
+    OUTPUT_DIR.mkdir(exist_ok=True)

cloudsql_to_supabase/utils.py

@@ -1,9 +1,55 @@
 import subprocess
+import shlex
+from typing import Dict, Optional, List, Union
+import logging
 
-def run_command(command, env=None):
-    print(f"Running command: {command}")
-    result = subprocess.run(command, shell=True, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-    if result.returncode != 0:
-        print(result.stderr.decode())
-        raise Exception(f"Command failed with error: {result.stderr.decode()}")
-    
+
+
+logger = logging.getLogger('cloud_to_supabase.utils')
+
+
+def run_command(cmd: str, env: Optional[Dict] = None, show_output: bool = True)-> subprocess.CompletedProcess:
+    """
+        Run a shell command safely with proper logging and error handling.
+    
+    Args:
+        cmd: Command to run
+        env: Environment variables
+        show_output: Whether to print command output to console
+        
+    Returns:
+        CompletedProcess instance
+    """
+    
+    safe_cmd = cmd
+    if "PGPASSWORD" in str(env):
+        safe_cmd = cmd.replace(env.get('PGPASSWORD', ''), '************')
+        logger.info(f'Running command: {safe_cmd}')
+        
+        
+        try: 
+            args = shlex.split(cmd)
+            result = subprocess.run(
+                args,
+                env=env,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                check=False
+            )
+            
+            if result.returncode != 0:
+                logger.error(f'command failed with exit code {result.returncode}')
+                logger.error(result.stderr)
+                raise RuntimeError(f'command failed: {result.stderr}')
+            
+            
+            if show_output and result.stdout:
+                logger.info(result.stdout)
+                
+                
+            return result
+        
+        except Exception as e:
+            logger.exception(f'Error executing command: {e}')
+            raise