Update typecheck err msg

sdks/python/apache_beam/io/gcp/pubsub_test.py

@@ -901,7 +901,8 @@ def test_write_messages_with_attributes_error(self, mock_pubsub):
 
     options = PipelineOptions([])
     options.view_as(StandardOptions).streaming = True
-    with self.assertRaisesRegex(Exception, r'Type hint violation'):
+    with self.assertRaisesRegex(Exception,
+                                r'requires.*PubsubMessage.*applied.*str'):
       with TestPipeline(options=options) as p:
         _ = (
             p

sdks/python/apache_beam/transforms/ptransform.py

@@ -497,13 +497,12 @@ def type_check_inputs_or_outputs(self, pvalueish, input_or_output):
         at_context = ' %s %s' % (input_or_output, context) if context else ''
         raise TypeCheckError(
             '{type} type hint violation at {label}{context}: expected {hint}, '
-            'got {actual_type}\nFull type hint:\n{debug_str}'.format(
+            'got {actual_type}'.format(
                 type=input_or_output.title(),
                 label=self.label,
                 context=at_context,
                 hint=hint,
-                actual_type=pvalue_.element_type,
-                debug_str=type_hints.debug_str()))
+                actual_type=pvalue_.element_type))
 
   def _infer_output_coder(self, input_type=None, input_coder=None):
     # type: (...) -> Optional[coders.Coder]
@@ -939,7 +938,21 @@ def element_type(side_input):
       bindings = getcallargs_forhints(argspec_fn, *arg_types, **kwargs_types)
       hints = getcallargs_forhints(
           argspec_fn, *input_types[0], **input_types[1])
-      for arg, hint in hints.items():
+      arg_hints = iter(hints.items())
+      element_arg, element_hint = next(arg_hints)
+      if not typehints.is_consistent_with(
+          bindings.get(element_arg, typehints.Any), element_hint):
+        transform_nest_level = self.label.count("/")
+        split_producer_label = pvalueish.producer.full_label.split("/")
+        producer_label = "/".join(
+            split_producer_label[:transform_nest_level + 1])
+        raise TypeCheckError(
+            f"The transform '{self.label}' requires "
+            f"PCollections of type '{element_hint}' "
+            f"but was applied to a PCollection of type"
+            f" '{bindings[element_arg]}' "
+            f"(produced by the transform '{producer_label}'). ")
+      for arg, hint in arg_hints:
         if arg.startswith('__unknown__'):
           continue
         if hint is None:

sdks/python/apache_beam/typehints/decorators_test.py

@@ -409,7 +409,7 @@ def fn(a: int) -> int:
       return a
 
     with self.assertRaisesRegex(TypeCheckError,
-                                r'requires .*int.* but got .*str'):
+                                r'requires .*int.* but was applied .*str'):
       _ = ['a', 'b', 'c'] | Map(fn)
 
     # Same pipeline doesn't raise without annotations on fn.
@@ -423,7 +423,7 @@ def fn(a: int) -> int:
     _ = [1, 2, 3] | Map(fn)  # Doesn't raise - correct types.
 
     with self.assertRaisesRegex(TypeCheckError,
-                                r'requires .*int.* but got .*str'):
+                                r'requires .*int.* but was applied .*str'):
       _ = ['a', 'b', 'c'] | Map(fn)
 
     @decorators.no_annotations

sdks/python/apache_beam/typehints/typed_pipeline_test.py

@@ -88,11 +88,11 @@ def process(self, element):
     self.assertEqual(['1', '2', '3'], sorted(result))
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'requires.*int.*got.*str'):
+                                r'requires.*int.*applied.*str'):
       ['a', 'b', 'c'] | beam.ParDo(MyDoFn())
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'requires.*int.*got.*str'):
+                                r'requires.*int.*applied.*str'):
       [1, 2, 3] | (beam.ParDo(MyDoFn()) | 'again' >> beam.ParDo(MyDoFn()))
 
   def test_typed_dofn_method(self):
@@ -104,11 +104,11 @@ def process(self, element: int) -> typehints.Tuple[str]:
     self.assertEqual(['1', '2', '3'], sorted(result))
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'requires.*int.*got.*str'):
+                                r'requires.*int.*applied.*str'):
       _ = ['a', 'b', 'c'] | beam.ParDo(MyDoFn())
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'requires.*int.*got.*str'):
+                                r'requires.*int.*applied.*str'):
       _ = [1, 2, 3] | (beam.ParDo(MyDoFn()) | 'again' >> beam.ParDo(MyDoFn()))
 
   def test_typed_dofn_method_with_class_decorators(self):
@@ -124,12 +124,12 @@ def process(self, element: int) -> typehints.Tuple[str]:
 
     with self.assertRaisesRegex(
         typehints.TypeCheckError,
-        r'requires.*Tuple\[<class \'int\'>, <class \'int\'>\].*got.*str'):
+        r'requires.*Tuple\[<class \'int\'>, <class \'int\'>\].*applied.*str'):
       _ = ['a', 'b', 'c'] | beam.ParDo(MyDoFn())
 
     with self.assertRaisesRegex(
         typehints.TypeCheckError,
-        r'requires.*Tuple\[<class \'int\'>, <class \'int\'>\].*got.*int'):
+        r'requires.*Tuple\[<class \'int\'>, <class \'int\'>\].*applied.*int'):
       _ = [1, 2, 3] | (beam.ParDo(MyDoFn()) | 'again' >> beam.ParDo(MyDoFn()))
 
   def test_typed_callable_iterable_output(self):
@@ -156,11 +156,11 @@ def process(self, element: typehints.Tuple[int, int]) -> \
     self.assertEqual(['1', '2', '3'], sorted(result))
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'requires.*int.*got.*str'):
+                                r'requires.*int.*applied.*str'):
       _ = ['a', 'b', 'c'] | beam.ParDo(my_do_fn)
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'requires.*int.*got.*str'):
+                                r'requires.*int.*applied.*str'):
       _ = [1, 2, 3] | (beam.ParDo(my_do_fn) | 'again' >> beam.ParDo(my_do_fn))
 
   def test_typed_callable_instance(self):
@@ -177,11 +177,11 @@ def do_fn(element: typehints.Tuple[int, int]) -> typehints.Generator[str]:
     self.assertEqual(['1', '2', '3'], sorted(result))
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'requires.*int.*got.*str'):
+                                r'requires.*int.*applied.*str'):
       _ = ['a', 'b', 'c'] | pardo
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'requires.*int.*got.*str'):
+                                r'requires.*int.*applied.*str'):
       _ = [1, 2, 3] | (pardo | 'again' >> pardo)
 
   def test_filter_type_hint(self):
@@ -430,7 +430,7 @@ def fn(element: float):
       return pcoll | beam.ParDo(fn)
 
     with self.assertRaisesRegex(typehints.TypeCheckError,
-                                r'ParDo.*requires.*float.*got.*str'):
+                                r'ParDo.*requires.*float.*applied.*str'):
       _ = ['1', '2', '3'] | MyMap()
     with self.assertRaisesRegex(typehints.TypeCheckError,
                                 r'MyMap.*expected.*str.*got.*bytes'):
@@ -632,14 +632,14 @@ def produces_unkown(e):
       return e
 
     @typehints.with_input_types(int)
-    def requires_int(e):
+    def accepts_int(e):
       return e
 
     class MyPTransform(beam.PTransform):
       def expand(self, pcoll):
         unknowns = pcoll | beam.Map(produces_unkown)
         ints = pcoll | beam.Map(int)
-        return (unknowns, ints) | beam.Flatten() | beam.Map(requires_int)
+        return (unknowns, ints) | beam.Flatten() | beam.Map(accepts_int)
 
     _ = [1, 2, 3] | MyPTransform()
 
@@ -761,8 +761,8 @@ def test_var_positional_only_side_input_hint(self):
 
     with self.assertRaisesRegex(
         typehints.TypeCheckError,
-        r'requires Tuple\[Union\[<class \'int\'>, <class \'str\'>\], ...\] but '
-        r'got Tuple\[Union\[<class \'float\'>, <class \'int\'>\], ...\]'):
+        r'requires.*Tuple\[Union\[<class \'int\'>, <class \'str\'>\], ...\].*'
+        r'applied.*Tuple\[Union\[<class \'float\'>, <class \'int\'>\], ...\]'):
       _ = [1.2] | beam.Map(lambda *_: 'a', 5).with_input_types(int, str)
 
   def test_var_keyword_side_input_hint(self):

sdks/python/pyproject.toml

@@ -26,7 +26,7 @@ requires = [
     # Avoid https://github.com/pypa/virtualenv/issues/2006
     "distlib==0.3.7",
     # Numpy headers
-    "numpy>=1.14.3,<2.2.0", # Update setup.py as well.
+    "numpy>=1.14.3,<1.27", # Update setup.py as well.
     # having cython here will create wheels that are platform dependent.
     "cython>=3.0,<4",
     ## deps for generating external transform wrappers: