问题描述
如何使用传递给 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
解决方法
我认为没有办法通过 python 的类型检查来强制执行此操作，尽管您始终可以添加自己的运行时类型检查以潜在地改进错误消息。
或者，您可以通过使用 Flex Templates 来完全避免使用 ValueProvider。
,我最终像这样创建了一个包装器值提供程序，欢迎提出意见/评论。
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,ValueProvider))
assert 'value' == dvp.get()