Introduce BaseExecutor.validate_command to avoid duplication (apache#…

…10033)

airflow/executors/base_executor.py

@@ -264,3 +264,9 @@ def terminate(self):
         This method is called when the daemon receives a SIGTERM
         """
         raise NotImplementedError()
+
+    @staticmethod
+    def validate_command(command: List[str]) -> None:
+        """Check if the command to execute is airflow comnand"""
+        if command[0:3] != ["airflow", "tasks", "run"]:
+            raise ValueError('The command must start with ["airflow", "tasks", "run"].')

airflow/executors/celery_executor.py

@@ -71,9 +71,7 @@
 @app.task
 def execute_command(command_to_exec: CommandType) -> None:
     """Executes command."""
-    if command_to_exec[0:3] != ["airflow", "tasks", "run"]:
-        raise ValueError('The command must start with ["airflow", "tasks", "run"].')
-
+    BaseExecutor.validate_command(command_to_exec)
     log.info("Executing command in Celery: %s", command_to_exec)
     env = os.environ.copy()
     try:

airflow/executors/dask_executor.py

@@ -72,8 +72,7 @@ def execute_async(self,
                       queue: Optional[str] = None,
                       executor_config: Optional[Any] = None) -> None:
 
-        if command[0:3] != ["airflow", "tasks", "run"]:
-            raise ValueError('The command must start with ["airflow", "tasks", "run"].')
+        self.validate_command(command)
 
         def airflow_run():
             return subprocess.check_call(command, close_fds=True)

airflow/executors/local_executor.py

@@ -287,8 +287,7 @@ def execute_async(self, key: TaskInstanceKey,
         if not self.impl:
             raise AirflowException(NOT_STARTED_MESSAGE)
 
-        if command[0:3] != ["airflow", "tasks", "run"]:
-            raise ValueError('The command must start with ["airflow", "tasks", "run"].')
+        self.validate_command(command)
 
         self.impl.execute_async(key=key, command=command, queue=queue, executor_config=executor_config)
 

airflow/executors/sequential_executor.py

@@ -49,10 +49,7 @@ def execute_async(self,
                       command: CommandType,
                       queue: Optional[str] = None,
                       executor_config: Optional[Any] = None) -> None:
-
-        if command[0:3] != ["airflow", "tasks", "run"]:
-            raise ValueError('The command must start with ["airflow", "tasks", "run"].')
-
+        self.validate_command(command)
         self.commands_to_run.append((key, command))
 
     def sync(self) -> None: