use task.run(context=context) where implemented + use optimistic patt…

…ern in CopyIncrementally

this will work around issues where task.run does not implement the context pattern e.g. RunFunction. These commands will always run locally

mara_pipelines/commands/sql.py

@@ -14,7 +14,7 @@
 from ..incremental_processing import file_dependencies
 from ..incremental_processing import incremental_copy_status
 from ..logging import logger
-from ..contexts import ExecutionContext, _LocalShellExecutionContext
+from ..contexts import ExecutionContext
 
 
 class _SQLCommand(pipelines.Command):
@@ -265,12 +265,7 @@ def target_db_alias(self):
         return self._target_db_alias or config.default_db_alias()
 
     def run(self, context: ExecutionContext = None) -> bool:
-        if isinstance(context, _LocalShellExecutionContext):
-            run_shell_command = context.run_shell_command
-        elif context is None:
-            run_shell_command = shell.run_shell_command
-        else:
-            raise ValueError('The context must inherit type _LocalShellExecutionContext')
+        run_shell_command = context.run_shell_command if context else shell.run_shell_command
 
         # retrieve the highest current value for the modification comparison (e.g.: the highest timestamp)
         # We intentionally use the command line here (rather than sqlalchemy) to avoid forcing people python drivers,

mara_pipelines/execution.py

@@ -443,8 +443,13 @@ def __init__(self, task: pipelines.Task, event_queue: multiprocessing.Queue, sta
         self.task = task
         self.event_queue = event_queue
         self.status_queue = status_queue
-        self.context = context
         self.start_time = datetime.datetime.now(tz.utc)
+        self.run_kargs = {}
+
+        # add dynamic kargs for self.task.run(...)
+        from inspect import signature
+        if 'context' in signature(task.run):
+            self.run_kargs['context'] = self.context
 
     def run(self):
         # redirect stdout and stderr to queue
@@ -454,7 +459,7 @@ def run(self):
         attempt = 0
         try:
             while True:
-                if not self.task.run(context=self.context):
+                if not self.task.run(**self.run_kargs):
                     max_retries = self.task.max_retries or config.default_task_max_retries()
                     if attempt < max_retries:
                         attempt += 1