Add exception chaining (#37423)

* Add exception chaining to preserve error context

- Add 'from e' to exception re-raises in CloudSQLEnrichmentHandler
- Add exception chaining in processes.py for OSError and CalledProcessError
- Improve logging in core.py to preserve traceback context

This improves debuggability by preserving the full exception chain,
following Python PEP 3134 best practices.

Fixes #37422

* Fix yapf formatting for logging.warning statement

* Fix yapf formatting: put logging arguments on single line

Author: shaheeramjad

CHANGES.md

@@ -68,6 +68,7 @@
 
 ## New Features / Improvements
 
+* (Python) Added exception chaining to preserve error context in CloudSQLEnrichmentHandler, processes utilities, and core transforms ([#37422](https://github.com/apache/beam/issues/37422)).
 * X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 
 ## Breaking Changes

sdks/python/apache_beam/transforms/core.py

@@ -2444,7 +2444,8 @@ def exception_handling_wrapper_do_fn_custom_process(self, *args, **kwargs):
         try:
           self._on_failure_callback(exn, args[0])
         except Exception as e:
-          logging.warning('on_failure_callback failed with error: %s', e)
+          logging.warning(
+              'on_failure_callback failed with error: %s', e, exc_info=True)
       yield pvalue.TaggedOutput(
           self._dead_letter_tag,
           (

sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql.py

@@ -395,11 +395,11 @@ def _execute_query(
         return data
       except Exception as e:
         transaction.rollback()
-        raise RuntimeError(f"Database operation failed: {e}")
+        raise RuntimeError(f"Database operation failed: {e}") from e
     except Exception as e:
       raise Exception(
           f'Could not execute the query. Please check if the query is properly '
-          f'formatted and the table exists. {e}')
+          f'formatted and the table exists. {e}') from e
     finally:
       if connection:
         connection.close()

sdks/python/apache_beam/utils/processes.py

@@ -49,56 +49,56 @@ def call(*args, **kwargs):
       kwargs['shell'] = True
     try:
       out = subprocess.call(*args, **kwargs)
-    except OSError:
-      raise RuntimeError("Executable {} not found".format(args[0]))
+    except OSError as e:
+      raise RuntimeError("Executable {} not found".format(args[0])) from e
     except subprocess.CalledProcessError as error:
       if isinstance(args, tuple) and (args[0][2] == "pip"):
         raise RuntimeError( \
           "Full traceback: {}\n Pip install failed for package: {} \
           \n Output from execution of subprocess: {}" \
-          .format(traceback.format_exc(), args[0][6], error. output))
+          .format(traceback.format_exc(), args[0][6], error. output)) from error
       else:
         raise RuntimeError("Full trace: {}\
            \n Output of the failed child process: {} " \
-          .format(traceback.format_exc(), error.output))
+          .format(traceback.format_exc(), error.output)) from error
     return out
 
   def check_call(*args, **kwargs):
     if force_shell:
       kwargs['shell'] = True
     try:
       out = subprocess.check_call(*args, **kwargs)
-    except OSError:
-      raise RuntimeError("Executable {} not found".format(args[0]))
+    except OSError as e:
+      raise RuntimeError("Executable {} not found".format(args[0])) from e
     except subprocess.CalledProcessError as error:
       if isinstance(args, tuple) and (args[0][2] == "pip"):
         raise RuntimeError( \
           "Full traceback: {} \n Pip install failed for package: {} \
           \n Output from execution of subprocess: {}" \
-          .format(traceback.format_exc(), args[0][6], error.output))
+          .format(traceback.format_exc(), args[0][6], error.output)) from error
       else:
         raise RuntimeError("Full trace: {} \
           \n Output of the failed child process: {}" \
-          .format(traceback.format_exc(), error.output))
+          .format(traceback.format_exc(), error.output)) from error
     return out
 
   def check_output(*args, **kwargs):
     if force_shell:
       kwargs['shell'] = True
     try:
       out = subprocess.check_output(*args, **kwargs)
-    except OSError:
-      raise RuntimeError("Executable {} not found".format(args[0]))
+    except OSError as e:
+      raise RuntimeError("Executable {} not found".format(args[0])) from e
     except subprocess.CalledProcessError as error:
       if isinstance(args, tuple) and (args[0][2] == "pip"):
         raise RuntimeError( \
           "Full traceback: {} \n Pip install failed for package: {} \
           \n Output from execution of subprocess: {}" \
-          .format(traceback.format_exc(), args[0][6], error.output))
+          .format(traceback.format_exc(), args[0][6], error.output)) from error
       else:
         raise RuntimeError("Full trace: {}, \
            output of the failed child process {} "\
-          .format(traceback.format_exc(), error.output))
+          .format(traceback.format_exc(), error.output)) from error
     return out
 
   def Popen(*args, **kwargs):