Add dependency support for delegating providers.

Author: robertwb

sdks/python/apache_beam/yaml/yaml_provider.py

@@ -138,10 +138,15 @@ def _affinity(self, other: "Provider"):
 
   @functools.cache
   def with_extra_dependencies(self, dependencies: Iterable[str]):
-    return self._with_extra_dependencies(dependencies)
+    result = self._with_extra_dependencies(dependencies)
+    if not hasattr(result, 'to_json'):
+      result.to_json = lambda: {'type': type(result).__name__}
+    return result
 
   def _with_extra_dependencies(self, dependencies: Iterable[str]):
-    raise ValueError('This provider does not support additional dependencies.')
+    raise ValueError(
+        'This provider of type %s does not support additional dependencies.' %
+        type(self).__name__)
 
 
 def as_provider(name, provider_or_constructor):
@@ -376,6 +381,8 @@ def is_path_or_urn(package):
             if is_path_or_urn(package) else package for package in packages
         ]))
 
+    self._packages = packages
+
   def available(self):
     return True  # If we're running this script, we have Python installed.
 
@@ -402,6 +409,10 @@ def _affinity(self, other: "Provider"):
     else:
       return super()._affinity(other)
 
+  def _with_extra_dependencies(self, dependencies: Iterable[str]):
+    return ExternalPythonProvider(
+        self._urns, None, set(self._packages) + set(dependencies))
+
 
 @ExternalProvider.register_provider_type('yaml')
 class YamlProvider(Provider):
@@ -1013,6 +1024,11 @@ def create_transform(
       yaml_create_transform: Any) -> beam.PTransform:
     return self._transforms[typ](self._underlying_provider, **config)
 
+  def _with_extra_dependencies(self, dependencies: Iterable[str]):
+    return TranslatingProvider(
+        self._transforms,
+        self._underlying_provider._with_extra_dependencies(dependencies))
+
 
 def create_java_builtin_provider():
   """Exposes built-in transforms from Java as well as Python to maximize
@@ -1292,6 +1308,14 @@ def underlying_provider(self):
   def cache_artifacts(self):
     self._underlying_provider.cache_artifacts()
 
+  def _with_extra_dependencies(self, dependencies: Iterable[str]):
+    return RenamingProvider(
+        self._transforms,
+        None,
+        self._mappings,
+        self._underlying_provider._with_extra_dependencies(dependencies),
+        self._defaults)
+
 
 def _as_list(func):
   @functools.wraps(func)