[GitHub] [beam] chamikaramj commented on a diff in pull request #17822: Allow creation of dynamically defined transforms in the Python expansion service.

[GitBox <gi...@apache.org>]

chamikaramj commented on code in PR #17822:
URL: https://github.com/apache/beam/pull/17822#discussion_r892949378


##########
sdks/python/apache_beam/transforms/fully_qualified_named_transform.py:
##########
@@ -52,16 +52,37 @@ def __init__(self, constructor, args, kwargs):
     self._kwargs = kwargs
 
   def expand(self, pinput):
-    return pinput | self._resolve(self._constructor)(
-        *self._args, **self._kwargs)
+    if self._constructor in ('__callable__', '__constructor__'):
+      self._check_allowed(self._constructor)
+      if self._args:
+        source, *args = tuple(self._args)
+        kwargs = self._kwargs
+      else:
+        args = self._args
+        kwargs = dict(self._kwargs)
+        source = kwargs.pop('source')
+
+      if self._constructor == '__constructor__':

Review Comment:
   So in the first case the function returns the PTransform. In the second case the function defines the "expand" method of a PTransform. Is that correct ?
   
   We should document all this properly in JavaExternalTransform since this usage is not clear without going through the details of the code.



##########
sdks/python/apache_beam/transforms/fully_qualified_named_transform.py:
##########
@@ -52,16 +52,37 @@ def __init__(self, constructor, args, kwargs):
     self._kwargs = kwargs
 
   def expand(self, pinput):
-    return pinput | self._resolve(self._constructor)(
-        *self._args, **self._kwargs)
+    if self._constructor in ('__callable__', '__constructor__'):

Review Comment:
   So the API is that user has to specific one of these strings as the method name and also make sure that the filter allows the same string ?



##########
sdks/python/apache_beam/utils/python_callable.py:
##########
@@ -88,6 +88,9 @@ def load_from_script(source):
       if line[0] != ' ':
         if line.startswith('def '):
           name = line[4:line.index('(')].strip()
+        elif line.startswith('class '):
+          name = line[5:line.index('(') if '(' in

Review Comment:
   Noting that we are extending the definition of the PythonCallable here, from a "function followed by some statements" to a "class followed by some statements that acts as a callable". This can be quite difficult to follow so please add appropriate documentation.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org