(beam) branch master updated: fix(sdk:python): Avoid AttributeError for transforms without hints (#36251)

[xq...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

xqhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 1e973630e67 fix(sdk:python): Avoid AttributeError for transforms without hints (#36251)
1e973630e67 is described below

commit 1e973630e6738a101e3840585d7d2832419018ce
Author: liferoad <hu...@gmail.com>
AuthorDate: Fri Oct 3 18:38:17 2025 -0400

    fix(sdk:python): Avoid AttributeError for transforms without hints (#36251)
    
    * fix(sdk:python): Avoid AttributeError for transforms without hints
    
    The AppliedPTransform initializer would unconditionally attempt to call
    `get_resource_hints()` on a transform object. This could cause an
    AttributeError if a PTransform implementation does not define this method.
    
    This change adds an `hasattr` check to verify the existence of the
    `get_resource_hints` method before calling it, preventing the potential
    crash and making the pipeline construction more robust.
    
    * fix annotations
    
    * fixed more
    
    * fixed more
    
    * more fixes
    
    * one more
---
 sdks/python/apache_beam/pipeline.py | 33 ++++++++++++++++++++++-----------
 1 file changed, 22 insertions(+), 11 deletions(-)

diff --git a/sdks/python/apache_beam/pipeline.py b/sdks/python/apache_beam/pipeline.py
index 02ce12a3bef..0e03c684153 100644
--- a/sdks/python/apache_beam/pipeline.py
+++ b/sdks/python/apache_beam/pipeline.py
@@ -354,7 +354,7 @@ class Pipeline(HasDisplayData):
           if replacement_transform is original_transform_node.transform:
             return
           replacement_transform.side_inputs = tuple(
-              original_transform_node.transform.side_inputs)
+              getattr(original_transform_node.transform, 'side_inputs', ()))
 
           replacement_transform_node = AppliedPTransform(
               original_transform_node.parent,
@@ -1027,7 +1027,9 @@ class Pipeline(HasDisplayData):
         # type: (AppliedPTransform) -> None
         if not transform_node.transform:
           return
-        if transform_node.transform.runner_api_requires_keyed_input():
+        if hasattr(
+            transform_node.transform, 'runner_api_requires_keyed_input'
+        ) and transform_node.transform.runner_api_requires_keyed_input():
           pcoll = transform_node.inputs[0]
           pcoll.element_type = typehints.coerce_to_kv_type(
               pcoll.element_type, transform_node.full_label)
@@ -1046,7 +1048,7 @@ class Pipeline(HasDisplayData):
                 == output.element_type.tuple_types[0]):
               output.requires_deterministic_key_coder = (
                   deterministic_key_coders and transform_node.full_label)
-        for side_input in transform_node.transform.side_inputs:
+        for side_input in getattr(transform_node.transform, 'side_inputs', []):
           if side_input.requires_keyed_input():
             side_input.pvalue.element_type = typehints.coerce_to_kv_type(
                 side_input.pvalue.element_type,
@@ -1240,10 +1242,10 @@ class AppliedPTransform(object):
     # once environment is a first-class citizen in Beam graph and we have
     # access to actual environment, not just an id.
     self.resource_hints = dict(
-        transform.get_resource_hints()) if transform else {
-        }  # type: Dict[str, bytes]
+        transform.get_resource_hints()) if transform and hasattr(
+            transform, 'get_resource_hints') else {}  # type: Dict[str, bytes]
 
-    if transform:
+    if transform and hasattr(transform, 'annotations'):
       annotations = {
           **(annotations or {}), **encode_annotations(transform.annotations())
       }
@@ -1399,8 +1401,11 @@ class AppliedPTransform(object):
       assert not self.main_inputs and not self.side_inputs
       return {}
     else:
-      named_inputs = self.transform._named_inputs(
-          self.main_inputs, self.side_inputs)
+      if hasattr(self.transform, '_named_inputs'):
+        named_inputs = self.transform._named_inputs(
+            self.main_inputs, self.side_inputs)
+      else:
+        named_inputs = {}
       if not self.parts:
         for name, pc_out in self.outputs.items():
           if pc_out.producer is not self and pc_out not in named_inputs.values(
@@ -1414,7 +1419,10 @@ class AppliedPTransform(object):
       assert not self.outputs
       return {}
     else:
-      return self.transform._named_outputs(self.outputs)
+      if hasattr(self.transform, '_named_outputs'):
+        return self.transform._named_outputs(self.outputs)
+      else:
+        return {}
 
   def to_runner_api(self, context):
     # type: (PipelineContext) -> beam_runner_api_pb2.PTransform
@@ -1441,7 +1449,9 @@ class AppliedPTransform(object):
               context,
               has_parts=bool(self.parts),
               named_inputs=self.named_inputs())
-        return transform.to_runner_api(context, has_parts=bool(self.parts))
+        elif hasattr(transform, 'to_runner_api'):
+          return transform.to_runner_api(context, has_parts=bool(self.parts))
+        return None
 
     # Iterate over inputs and outputs by sorted key order, so that ids are
     # consistently generated for multiple runs of the same pipeline.
@@ -1527,7 +1537,8 @@ class AppliedPTransform(object):
         environment_id=None,
         annotations=proto.annotations)
 
-    if result.transform and result.transform.side_inputs:
+    if result.transform and hasattr(
+        result.transform, 'side_inputs') and result.transform.side_inputs:
       for si, pcoll in zip(result.transform.side_inputs, side_inputs):
         si.pvalue = pcoll
       result.side_inputs = tuple(result.transform.side_inputs)