damccorm commented on code in PR #35965:
URL: https://github.com/apache/beam/pull/35965#discussion_r2302140605
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2354,14 +2356,28 @@ def __init__(
self._partial = partial
self._on_failure_callback = on_failure_callback
+ # Wrap process and expose any top level state params so that process can
+ # handle state.
+ @wraps(self._fn.process)
+ def process_wrapper(*args, **kwargs):
+ return self.custom_process(*args, **kwargs)
+ self.process = process_wrapper
+ for i in dir(self._fn):
+ member = getattr(self._fn, i)
+ if isinstance(member, StateSpec):
+ setattr(self, i, member)
+
+ from apache_beam.runners.common import DoFnSignature
+ raise ValueError('is stateful: ' +
str(DoFnSignature(self).is_stateful_dofn()) +
str(inspect.signature(self.process)))
Review Comment:
This is returning that it is not stateful, but it is returning the expected
process signature. Something is a little off here.
--
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-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
