jrmccluskey commented on issue #33189:
URL: https://github.com/apache/beam/issues/33189#issuecomment-2498656947
I dug into this a little bit, and we can get to a functional pipeline with a
few small tweaks:
```
import apache_beam as beam
from typing import Generic
from typing import TypeVar
from apache_beam.testing.util import assert_that, equal_to
from apache_beam.testing.test_pipeline import TestPipeline
T = TypeVar("T")
class ResultContainer(Generic[T]):
def __init__(self, payload: T) -> None:
self.payload = payload
class DifferentDoFn(beam.DoFn):
def process(self, data: int):
yield ResultContainer[int](data)
class SomeDoFn(beam.DoFn):
def process(self, data: ResultContainer[int]):
yield data.payload + 1
def test_pardo():
with TestPipeline() as p:
output = (p | beam.Create([1])
| beam.ParDo(DifferentDoFn())
| beam.ParDo(SomeDoFn()))
assert_that(
label="check result",
actual=output,
matcher=equal_to([2]),
)
if __name__ == '__main__':
test_pardo()
```
The problem the type hinting infrastructure runs into is that creating the
subscripted generic type directly in the `Create` function leads to some funky
comparisons in our type checking code. We were comparing a generic type
(`ResultContainer(1)` not carrying a type at all) and the type
`ResultContainer[int]`. Regardless of the former not having a type, Python's
built-in comparison tools for `isinstance()` and `issubclass()` don't like
these generics, hence the error message.
In the short term, providing a pattern like this where you add an extra DoFn
to construct the types should be a valid workaround, in the longer term this
problem is a little tricky but seems fixable with the help of some third-party
type introspection code that I've started tinkering with.
--
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]
