Enable ruff PLW1509 rule

providers/standard/src/airflow/providers/standard/hooks/subprocess.py

@@ -77,14 +77,6 @@ def run_command(
         """
         self.log.info("Tmp dir root location: %s", gettempdir())
         with working_directory(cwd=cwd) as cwd:
-
-            def pre_exec():
-                # Restore default signal disposition and invoke setsid
-                for sig in ("SIGPIPE", "SIGXFZ", "SIGXFSZ"):
-                    if hasattr(signal, sig):
-                        signal.signal(getattr(signal, sig), signal.SIG_DFL)
-                os.setsid()
-
             self.log.info("Running command: %s", command)
 
             self.sub_process = Popen(
@@ -93,7 +85,8 @@ def pre_exec():
                 stderr=STDOUT,
                 cwd=cwd,
                 env=env if env or env == {} else os.environ,
-                preexec_fn=pre_exec,
+                start_new_session=True,
+                restore_signals=True,
             )
 
             self.log.info("Output:")

providers/standard/src/airflow/providers/standard/sensors/bash.py

@@ -17,7 +17,6 @@
 # under the License.
 from __future__ import annotations
 
-import os
 from collections.abc import Sequence
 from subprocess import PIPE, STDOUT, Popen
 from tempfile import NamedTemporaryFile, TemporaryDirectory, gettempdir
@@ -89,7 +88,7 @@ def poke(self, context: Context):
                 close_fds=True,
                 cwd=tmp_dir,
                 env=self.env,
-                preexec_fn=os.setsid,
+                start_new_session=True,
             ) as resp:
                 if resp.stdout:
                     self.log.info("Output:")

providers/standard/tests/unit/standard/hooks/test_subprocess.py

@@ -97,7 +97,8 @@ def test_should_exec_subprocess(self, mock_popen, mock_temporary_directory):
             ["bash", "-c", 'echo "stdout"'],
             cwd="/tmp/airflowtmpcatcat",
             env={},
-            preexec_fn=mock.ANY,
+            restore_signals=True,
+            start_new_session=True,
             stderr=STDOUT,
             stdout=PIPE,
         )

providers/teradata/src/airflow/providers/teradata/hooks/bteq.py

@@ -263,7 +263,7 @@ def execute_bteq_script_at_local(
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
             shell=True,
-            preexec_fn=os.setsid,
+            start_new_session=True,
         )
         encode_bteq_script = bteq_script.encode(str(temp_file_read_encoding or "UTF-8"))
         stdout_data, _ = process.communicate(input=encode_bteq_script)

providers/teradata/tests/unit/teradata/hooks/test_bteq.py

@@ -150,7 +150,7 @@ def test_execute_bteq_script_at_local_success(
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
         shell=True,
-        preexec_fn=os.setsid,
+        start_new_session=True,
     )
     assert ret_code == 0
 

pyproject.toml

@@ -610,6 +610,7 @@ extend-select = [
     "PLW1501", # {mode} is not a valid mode for open
     "PLW1507", # Shallow copy of os.environ via copy.copy(os.environ)
     "PLW1508", # Invalid type for environment variable default; expected str or None
+    "PLW1509", # preexec_fn argument is unsafe when using threads
     "PLW1510", # subprocess.run without explicit check argument
     "PLW1641", # Object does not implement __hash__ method
     # Per rule enables