Python Apache beam 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

我认为没有办法通过 python 的类型检查来强制执行此操作，尽管您始终可以添加自己的运行时类型检查以潜在地改进错误消息。

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

Answer 2

我最终像这样创建了一个包装器值提供程序，opinions/comments 欢迎。

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()

Python Apache beam ValueProvider 的类型提示

Python Type Hints for Apache beam ValueProvider

apache-beam

apache-beam-io