fix import errors

cgatcore/pipeline/base_executor.py

@@ -68,3 +68,52 @@ def __exit__(self, exc_type, exc_value, traceback):
         """Exit the runtime context related to this object."""
         # Cleanup logic, if any, can be added here
         pass
+
+
+class Executor(BaseExecutor):
+    """Main executor class that handles job execution and resource management."""
+
+    def __init__(self, **kwargs):
+        """Initialize with configuration options."""
+        super().__init__(**kwargs)
+        self.task_name = "executor_task"
+        self.default_total_time = 5
+
+        # Initialize job options
+        self.job_options = kwargs.get('job_options', '')
+        self.queue = kwargs.get('queue')
+        self.cluster_queue_manager = kwargs.get('cluster_queue_manager', 'slurm')
+
+    def run(self, statement_list, **kwargs):
+        """Execute a list of statements.
+        
+        Args:
+            statement_list (list): List of commands to execute
+            **kwargs: Additional execution options
+            
+        Returns:
+            tuple: (exit_code, stdout, stderr)
+        """
+        if isinstance(statement_list, str):
+            statement_list = [statement_list]
+
+        results = []
+        for statement in statement_list:
+            # Choose appropriate executor based on configuration
+            if self.cluster_queue_manager == 'slurm':
+                from cgatcore.pipeline.executors import SlurmExecutor
+                executor = SlurmExecutor(**self.config)
+            elif self.cluster_queue_manager == 'sge':
+                from cgatcore.pipeline.executors import SGEExecutor
+                executor = SGEExecutor(**self.config)
+            elif self.cluster_queue_manager == 'torque':
+                from cgatcore.pipeline.executors import TorqueExecutor
+                executor = TorqueExecutor(**self.config)
+            else:
+                from cgatcore.pipeline.executors import LocalExecutor
+                executor = LocalExecutor(**self.config)
+
+            result = executor.run(statement)
+            results.append(result)
+
+        return results[0] if len(results) == 1 else results

cgatcore/pipeline/control.py

@@ -49,10 +49,11 @@
 from cgatcore.pipeline.utils import get_caller, get_caller_locals, is_test
 from cgatcore.pipeline.execution import (
     execute, start_session, get_drmaa_session, close_session)
+from cgatcore.pipeline.executors import (
+    Executor, LocalExecutor, SlurmExecutor, SGEExecutor, TorqueExecutor)
 
 # redirect os.stat and other OS utilities to cached versions to speed
 # up ruffus. Be careful not to use os.stat in task functions.
-# Does ruffus rely on os.stat for completed tasks?
 SAVED_OS_STAT = os.stat
 SAVED_OS_PATH_ABSPATH = os.path.abspath
 SAVED_OS_PATH_REALPATH = os.path.realpath