Apache束的Python类型提示ValueProvider

Question

如何使用传递给PTransform和DoFn类的ValueProvider值类型的类型提示？

class MyPTransform(beam.PTransform):
  def __init__(self, my_value_provider: ValueProvider):
     # How do I enforce my_value_provider has value_type of str
     self.my_value_provider = my_value_provider

我可以将其设置为RuntimeValueProvider或StaticValueProvider并显式测试：

 type(my_value_provider.type) == str

其他人是如何做到这一点的？我在这里什么也没看到：https://beam.apache.org/documentation/sdks/python-type-safety

Answer 1

我最终创建了一个包装器值提供者，如下所示，欢迎发表意见/评论。

from typing import Generic, TypeVar

from apache_beam.options.value_provider import ValueProvider, StaticValueProvider, RuntimeValueProvider

T = TypeVar('T')


class TypedValueProvider(ValueProvider, Generic[T]):
    """
    Providers a wrapper around beam's ValueProvider to allow specifying value type hints
    """

    def __init__(self, value_provider: ValueProvider, value_type: Type[T]):
        self.value_provider = value_provider
        self.value_type = value_type
        assert value_provider.value_type == value_type

    def is_accessible(self):
        return self.value_provider.is_accessible()

    def get(self):
        return self.value_provider.get()


if __name__ == '__main__':
    svp = StaticValueProvider(str, 'blah')
    dvp: TypedValueProvider[str] = TypedValueProvider(svp, str)
    print(isinstance(dvp, ValueProvider))
    assert 'blah' == dvp.get()

    rvp = RuntimeValueProvider('key', str, 'default')
    RuntimeValueProvider.set_runtime_options({'key': 'value'})

    dvp: TypedValueProvider[str] = TypedValueProvider(rvp, str)
    print(isinstance(dvp, ValueProvider))
    assert 'value' == dvp.get()

Answer 2

我不认为有办法通过python的类型检查来强制执行这一点，尽管您可以随时添加自己的运行时类型检查来潜在地改进错误消息。

或者，您可以通过使用Flex模板来避免使用ValueProvider。