claudevdm commented on code in PR #37522:
URL: https://github.com/apache/beam/pull/37522#discussion_r2790936805
##########
sdks/python/apache_beam/pipeline.py:
##########
@@ -698,6 +705,15 @@ def apply(
RuntimeError: if the transform object was already applied to
this pipeline and needs to be cloned in order to apply again.
"""
+ with scoped_pipeline_options(self._options):
+ return self._apply_internal(transform, pvalueish, label)
+
+ def _apply_internal(
+ self,
+ transform: ptransform.PTransform,
+ pvalueish: Optional[pvalue.PValue] = None,
+ label: Optional[str] = None) -> pvalue.PValue:
+ """Internal implementation of apply(), called within scoped options."""
if isinstance(transform, ptransform._NamedPTransform):
return self.apply(
transform.transform, pvalueish, label or transform.label)
Review Comment:
Yes it is allowed and handles recursion with set/reset. I thought about
guarding entering the context manager in apply with
```
if self._options is get_scoped_pipeline_options():
return self.apply_internal(
with scoped_pipeline_options(self.options):
return self.apply_internal(
```
I decided against it because context var already behaves as expected and is
efficient. WDYT?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
