[3/4] incubator-beam git commit: Better error message for poor use of callable apply

[ro...@apache.org]

Better error message for poor use of callable apply


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c34f332a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c34f332a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c34f332a

Branch: refs/heads/python-sdk
Commit: c34f332a3ae6e4ad914965732f6a038a883a5b3b
Parents: 31b3f00
Author: Robert Bradshaw <ro...@google.com>
Authored: Wed Jul 6 15:42:55 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Thu Jul 7 11:50:50 2016 -0700

----------------------------------------------------------------------
 sdks/python/apache_beam/pipeline.py      |  4 ++++
 sdks/python/apache_beam/pipeline_test.py | 13 ++++++++++++-
 2 files changed, 16 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c34f332a/sdks/python/apache_beam/pipeline.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pipeline.py b/sdks/python/apache_beam/pipeline.py
index a84cec3..012d4d9 100644
--- a/sdks/python/apache_beam/pipeline.py
+++ b/sdks/python/apache_beam/pipeline.py
@@ -47,6 +47,7 @@ import logging
 import os
 import shutil
 import tempfile
+import types
 
 from apache_beam import pvalue
 from apache_beam import typehints
@@ -196,6 +197,9 @@ class Pipeline(object):
         and needs to be cloned in order to apply again.
     """
     if not isinstance(transform, ptransform.PTransform):
+      if isinstance(transform, (type, types.ClassType)):
+        raise TypeError("%s is not a PTransform instance, did you mean %s()?"
+                        % (transform, transform.__name__))
       transform = _CallableWrapperPTransform(transform)
 
     full_label = format_full_label(self._current_transform(), transform)

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c34f332a/sdks/python/apache_beam/pipeline_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pipeline_test.py b/sdks/python/apache_beam/pipeline_test.py
index 5e94087..8598737 100644
--- a/sdks/python/apache_beam/pipeline_test.py
+++ b/sdks/python/apache_beam/pipeline_test.py
@@ -32,6 +32,7 @@ from apache_beam.transforms import Create
 from apache_beam.transforms import FlatMap
 from apache_beam.transforms import Flatten
 from apache_beam.transforms import Map
+from apache_beam.transforms import GroupByKey
 from apache_beam.transforms import PTransform
 from apache_beam.transforms import Read
 from apache_beam.transforms.util import assert_that, equal_to
@@ -174,10 +175,20 @@ class PipelineTest(unittest.TestCase):
   def test_apply_custom_callable(self):
     pipeline = Pipeline(self.runner_name)
     pcoll = pipeline | Create('pcoll', [1, 2, 3])
-    result = pipeline.apply(PipelineTest.custom_callable, pcoll)
+    result = pcoll | PipelineTest.custom_callable
     assert_that(result, equal_to([2, 3, 4]))
     pipeline.run()
 
+  def test_apply_custom_callable_error(self):
+    pipeline = Pipeline(self.runner_name)
+    pcoll = pipeline | Create('pcoll', [1, 2, 3])
+    with self.assertRaises(TypeError) as cm:
+      pcoll | GroupByKey  # Note the missing ()'s
+    self.assertEqual(
+        cm.exception.message,
+        "<class 'apache_beam.transforms.core.GroupByKey'> is not "
+        "a PTransform instance, did you mean GroupByKey()?")
+
   def test_transform_no_super_init(self):
     class AddSuffix(PTransform):