From d23e964b70e44aab9bb29ba803d335ab971f77df Mon Sep 17 00:00:00 2001 From: GAURAV THAKUR Date: Fri, 16 Oct 2026 07:38:58 +0530 Subject: [PATCH 1/4] [BEAM-19954] Resolve string annotations in callable type hints Use typing.get_type_hints for annotated functions and class constructors so forward references are converted to their intended Beam type hints. Preserve the existing Any fallback when annotations cannot be resolved. Add regression coverage for string and future annotations, methods, constructors, and unresolved names. --- .../apache_beam/typehints/decorators.py | 20 +++++++++ .../apache_beam/typehints/decorators_test.py | 45 +++++++++++++++++++ 2 files changed, 65 insertions(+) diff --git a/sdks/python/apache_beam/typehints/decorators.py b/sdks/python/apache_beam/typehints/decorators.py index bc44ec745d5a..d6860c5f39c1 100644 --- a/sdks/python/apache_beam/typehints/decorators.py +++ b/sdks/python/apache_beam/typehints/decorators.py @@ -85,6 +85,7 @@ def foo((a, b)): import logging import traceback import types +import typing from typing import Any from typing import Callable from typing import Dict @@ -334,6 +335,25 @@ def from_callable(cls, fn: Callable) -> Optional['IOTypeHints']: for param in signature.parameters.values()) and signature.return_annotation == signature.empty): return None + if (any(isinstance(param.annotation, str) + for param in signature.parameters.values()) or + isinstance(signature.return_annotation, str)): + try: + resolved = typing.get_type_hints( + fn.__init__ if inspect.isclass(fn) else fn) + except (AttributeError, NameError, SyntaxError, TypeError): + # Keep unresolved annotations on the existing Any fallback path. + resolved = {} + signature = signature.replace( + parameters=[ + param.replace( + annotation=resolved.get(param.name, param.annotation)) + if isinstance(param.annotation, str) else param + for param in signature.parameters.values() + ], + return_annotation=(resolved.get('return', signature.return_annotation) + if isinstance(signature.return_annotation, str) else + signature.return_annotation)) input_args = [] input_kwargs = {} for param in signature.parameters.values(): diff --git a/sdks/python/apache_beam/typehints/decorators_test.py b/sdks/python/apache_beam/typehints/decorators_test.py index 73efea9e9abf..d5901c9aa2a8 100644 --- a/sdks/python/apache_beam/typehints/decorators_test.py +++ b/sdks/python/apache_beam/typehints/decorators_test.py @@ -256,6 +256,51 @@ def fn( })) self.assertEqual(th.output_types, ((Tuple[Any, ...], ), {})) + def test_from_callable_resolves_string_annotations(self): + def fn(value: 'int') -> 'typehints.List[str]': + return [str(value)] + + th = decorators.IOTypeHints.from_callable(fn) + self.assertEqual(th.input_types, ((int, ), {})) + self.assertEqual(th.output_types, ((List[str], ), {})) + + def test_from_callable_resolves_future_annotations(self): + namespace = {'typehints': typehints} + exec( + 'from __future__ import annotations\n' + 'def fn(value: int) -> typehints.List[str]:\n' + ' return [str(value)]\n', namespace) + + th = decorators.IOTypeHints.from_callable(namespace['fn']) + self.assertEqual(th.input_types, ((int, ), {})) + self.assertEqual(th.output_types, ((List[str], ), {})) + + def test_from_callable_resolves_class_constructor_string_annotations(self): + class Class(object): + def __init__(self, value: 'int'): + self.value = value + + th = decorators.IOTypeHints.from_callable(Class) + self.assertEqual(th.input_types, ((int, ), {})) + self.assertEqual(th.output_types, ((Class, ), {})) + + def test_from_callable_resolves_method_string_annotations(self): + class Class(object): + def process(self, value: 'int') -> 'typehints.List[str]': + return [str(value)] + + th = decorators.IOTypeHints.from_callable(Class().process) + self.assertEqual(th.input_types, ((int, ), {})) + self.assertEqual(th.output_types, ((List[str], ), {})) + + def test_from_callable_unresolved_string_annotations_fall_back_to_any(self): + def fn(value: 'MissingType') -> 'int': + return 1 + + th = decorators.IOTypeHints.from_callable(fn) + self.assertEqual(th.input_types, ((Any, ), {})) + self.assertEqual(th.output_types, ((Any, ), {})) + def test_from_callable_partial(self): def fn(a: int) -> int: return a From 2f72af338743e2bb530962861310adabb0a2394b Mon Sep 17 00:00:00 2001 From: GAURAV THAKUR Date: Fri, 16 Oct 2026 07:55:25 +0530 Subject: [PATCH 2/4] Apply Beam YAPF formatting to forward-reference fix Match the formatter's expected layout for the return annotation and postponed-annotation regression test. --- sdks/python/apache_beam/typehints/decorators.py | 7 ++++--- sdks/python/apache_beam/typehints/decorators_test.py | 3 ++- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/sdks/python/apache_beam/typehints/decorators.py b/sdks/python/apache_beam/typehints/decorators.py index d6860c5f39c1..2ae8c0dbf11f 100644 --- a/sdks/python/apache_beam/typehints/decorators.py +++ b/sdks/python/apache_beam/typehints/decorators.py @@ -351,9 +351,10 @@ def from_callable(cls, fn: Callable) -> Optional['IOTypeHints']: if isinstance(param.annotation, str) else param for param in signature.parameters.values() ], - return_annotation=(resolved.get('return', signature.return_annotation) - if isinstance(signature.return_annotation, str) else - signature.return_annotation)) + return_annotation=( + resolved.get('return', signature.return_annotation) if isinstance( + signature.return_annotation, str) else + signature.return_annotation)) input_args = [] input_kwargs = {} for param in signature.parameters.values(): diff --git a/sdks/python/apache_beam/typehints/decorators_test.py b/sdks/python/apache_beam/typehints/decorators_test.py index d5901c9aa2a8..f7752d362730 100644 --- a/sdks/python/apache_beam/typehints/decorators_test.py +++ b/sdks/python/apache_beam/typehints/decorators_test.py @@ -269,7 +269,8 @@ def test_from_callable_resolves_future_annotations(self): exec( 'from __future__ import annotations\n' 'def fn(value: int) -> typehints.List[str]:\n' - ' return [str(value)]\n', namespace) + ' return [str(value)]\n', + namespace) th = decorators.IOTypeHints.from_callable(namespace['fn']) self.assertEqual(th.input_types, ((int, ), {})) From 8e46db42cb7a93d916a35fe3f62da335b376edf9 Mon Sep 17 00:00:00 2001 From: GAURAV THAKUR Date: Fri, 16 Oct 2026 09:24:35 +0530 Subject: [PATCH 3/4] [BEAM-19954] Resolve forward references independently and correct ParDo hint coverage --- .../apache_beam/typehints/decorators.py | 83 ++++++++-- .../apache_beam/typehints/decorators_test.py | 145 +++++++++++++++++- .../typehints/typed_pipeline_test.py | 5 +- 3 files changed, 216 insertions(+), 17 deletions(-) diff --git a/sdks/python/apache_beam/typehints/decorators.py b/sdks/python/apache_beam/typehints/decorators.py index 2ae8c0dbf11f..d3547f535681 100644 --- a/sdks/python/apache_beam/typehints/decorators.py +++ b/sdks/python/apache_beam/typehints/decorators.py @@ -80,6 +80,7 @@ def foo((a, b)): # pytype: skip-file # ruff: noqa: UP006 +import functools import inspect import itertools import logging @@ -132,6 +133,67 @@ def foo((a, b)): _disable_from_callable = False # pylint: disable=invalid-name +def _contains_forward_ref(annotation): + if isinstance(annotation, (str, typing.ForwardRef)): + return True + origin = get_origin(annotation) + if origin is Literal: + # Literal strings are values, not type names. + return False + args = get_args(annotation) + if origin is typing.Annotated: + args = args[:1] # The remaining arguments are metadata. + return any(_contains_forward_ref(arg) for arg in args) + + +def _annotation_namespaces(fn): + """Find the implementation and namespaces behind inspect.signature(fn).""" + target = fn + while isinstance(target, functools.partial): + target = target.func + localns = None + if inspect.isclass(target): + localns = dict(vars(target)) + localns[target.__name__] = target + if target.__init__ is not object.__init__: + target = target.__init__ + else: + target = target.__new__ + elif inspect.ismethod(target): + owner = target.__self__ + owner = owner if inspect.isclass(owner) else type(owner) + localns = dict(vars(owner)) + localns[owner.__name__] = owner + elif not (inspect.isfunction(target) or inspect.isbuiltin(target)): + owner = type(target) + target = target.__call__ + localns = dict(vars(owner)) + localns[owner.__name__] = owner + target = inspect.unwrap(target) + if inspect.ismethod(target): + target = target.__func__ + return getattr(target, '__globals__', {}), localns + + +def _resolve_annotation(annotation, globalns, localns): + if not _contains_forward_ref(annotation): + return annotation + + # get_type_hints resolves nested ForwardRefs and uses public typing APIs. + # A one-hint callable lets an unresolved name leave other hints intact. + def one_hint(): + pass + + one_hint.__annotations__ = {'value': annotation} + try: + return typing.get_type_hints( + one_hint, globalns=globalns, localns=localns, + include_extras=True)['value'] + except (AttributeError, NameError, SyntaxError, TypeError): + # convert_to_beam_type maps unresolved strings and ForwardRefs to Any. + return annotation + + def get_signature(func): """Like inspect.signature(), but supports Py2 as well. @@ -335,26 +397,19 @@ def from_callable(cls, fn: Callable) -> Optional['IOTypeHints']: for param in signature.parameters.values()) and signature.return_annotation == signature.empty): return None - if (any(isinstance(param.annotation, str) + if (any(_contains_forward_ref(param.annotation) for param in signature.parameters.values()) or - isinstance(signature.return_annotation, str)): - try: - resolved = typing.get_type_hints( - fn.__init__ if inspect.isclass(fn) else fn) - except (AttributeError, NameError, SyntaxError, TypeError): - # Keep unresolved annotations on the existing Any fallback path. - resolved = {} + _contains_forward_ref(signature.return_annotation)): + globalns, localns = _annotation_namespaces(fn) signature = signature.replace( parameters=[ param.replace( - annotation=resolved.get(param.name, param.annotation)) - if isinstance(param.annotation, str) else param + annotation=_resolve_annotation( + param.annotation, globalns, localns)) for param in signature.parameters.values() ], - return_annotation=( - resolved.get('return', signature.return_annotation) if isinstance( - signature.return_annotation, str) else - signature.return_annotation)) + return_annotation=_resolve_annotation( + signature.return_annotation, globalns, localns)) input_args = [] input_kwargs = {} for param in signature.parameters.values(): diff --git a/sdks/python/apache_beam/typehints/decorators_test.py b/sdks/python/apache_beam/typehints/decorators_test.py index f7752d362730..bcff8673479d 100644 --- a/sdks/python/apache_beam/typehints/decorators_test.py +++ b/sdks/python/apache_beam/typehints/decorators_test.py @@ -264,6 +264,14 @@ def fn(value: 'int') -> 'typehints.List[str]': self.assertEqual(th.input_types, ((int, ), {})) self.assertEqual(th.output_types, ((List[str], ), {})) + def test_from_callable_pardo_output_is_an_iterable_of_elements(self): + def do_fn(element: 'int') -> 'typehints.List[typehints.List[str]]': + return [[str(element)] * 2] + + th = decorators.IOTypeHints.from_callable(do_fn) + self.assertEqual(th.input_types, ((int, ), {})) + self.assertEqual(th.strip_iterable().output_types, ((List[str], ), {})) + def test_from_callable_resolves_future_annotations(self): namespace = {'typehints': typehints} exec( @@ -300,7 +308,142 @@ def fn(value: 'MissingType') -> 'int': th = decorators.IOTypeHints.from_callable(fn) self.assertEqual(th.input_types, ((Any, ), {})) - self.assertEqual(th.output_types, ((Any, ), {})) + self.assertEqual(th.output_types, ((int, ), {})) + + def test_from_callable_resolves_other_hints_after_missing_name(self): + namespace = {} + exec( + 'from __future__ import annotations\n' + 'def fn(missing: MissingType, value: int, *, name: str) -> int:\n' + ' return value\n', + namespace) + + th = decorators.IOTypeHints.from_callable(namespace['fn']) + self.assertEqual(th.input_types, ((Any, int), {'name': str})) + self.assertEqual(th.output_types, ((int, ), {})) + + def test_from_callable_resolves_nested_forward_references(self): + def fn( + values: typing.List['int'], + pair: typing.Tuple['str', int], + builtin: list['int'], + nested: typing.Dict[str, typing.List[typing.ForwardRef('int')]], + resolved: typing.List[int]) -> typing.List['str']: + return [str(values[0])] + + th = decorators.IOTypeHints.from_callable(fn) + self.assertEqual( + th.input_types, + (( + List[int], + Tuple[str, int], + List[int], + Dict[str, List[int]], + List[int]), {})) + self.assertEqual(th.output_types, ((List[str], ), {})) + + def test_from_callable_unresolved_nested_reference_falls_back_to_any(self): + def fn(values: typing.List['MissingType']) -> 'int': + return len(values) + + th = decorators.IOTypeHints.from_callable(fn) + self.assertEqual(th.input_types, ((List[Any], ), {})) + self.assertEqual(th.output_types, ((int, ), {})) + + def test_from_callable_partial_with_string_annotations(self): + def fn(bound: 'int', value: 'str', *, option: 'int') -> 'str': + return value + + th = decorators.IOTypeHints.from_callable(functools.partial(fn, 1)) + self.assertEqual(th.input_types, ((str, ), { + 'option': int + })) + self.assertEqual(th.output_types, ((str, ), {})) + + def test_from_callable_bound_and_unbound_methods_with_strings(self): + class Processor: + def process(self, value: 'int', *, label: 'str') -> 'str': + return label + + unbound = decorators.IOTypeHints.from_callable(Processor.process) + self.assertEqual(unbound.input_types, ((Any, int), {'label': str})) + self.assertEqual(unbound.output_types, ((str, ), {})) + bound = decorators.IOTypeHints.from_callable(Processor().process) + self.assertEqual(bound.input_types, ((int, ), { + 'label': str + })) + self.assertEqual(bound.output_types, ((str, ), {})) + + def test_from_callable_varargs_and_kwargs_with_strings(self): + def fn( + value: 'int', + *args: 'typing.Tuple[str]', + required: 'str', + **kwargs: 'typing.Dict[str, int]') -> 'int': + return value + + th = decorators.IOTypeHints.from_callable(fn) + self.assertEqual( + th.input_types, ((int, Tuple[str]), { + 'required': str, 'kwargs': Dict[str, int] + })) + self.assertEqual(th.output_types, ((int, ), {})) + + def test_from_callable_callable_instance_with_string_annotations(self): + class CallableClass: + def __call__(self, value: 'int') -> 'str': + return str(value) + + th = decorators.IOTypeHints.from_callable(CallableClass()) + self.assertEqual(th.input_types, ((int, ), {})) + self.assertEqual(th.output_types, ((str, ), {})) + + def test_from_callable_wrapped_function_uses_original_namespace(self): + namespace = {'Alias': str} + exec('def original(value: "Alias") -> "int":\n return 1\n', namespace) + + @functools.wraps(namespace['original']) + def wrapper(*args, **kwargs): + return namespace['original'](*args, **kwargs) + + th = decorators.IOTypeHints.from_callable(wrapper) + self.assertEqual(th.input_types, ((str, ), {})) + self.assertEqual(th.output_types, ((int, ), {})) + + def test_from_callable_class_new_and_inherited_init(self): + class WithNew: + def __new__(cls, value: 'int'): + return super().__new__(cls) + + class NoConstructor: + pass + + th = decorators.IOTypeHints.from_callable(WithNew) + self.assertEqual(th.input_types, ((int, ), {})) + self.assertEqual(th.output_types, ((WithNew, ), {})) + th = decorators.IOTypeHints.from_callable(NoConstructor) + self.assertEqual(th.input_types, ((), {})) + self.assertEqual(th.output_types, ((NoConstructor, ), {})) + + def test_from_callable_resolves_class_local_names(self): + class Node: + def __init__(self, previous: 'Node'): + self.previous = previous + + def link(self, next_node: 'Node') -> 'Node': + return next_node + + th = decorators.IOTypeHints.from_callable(Node) + self.assertEqual(th.input_types, ((Node, ), {})) + self.assertEqual(th.output_types, ((Node, ), {})) + th = decorators.IOTypeHints.from_callable(Node(None).link) + self.assertEqual(th.input_types, ((Node, ), {})) + self.assertEqual(th.output_types, ((Node, ), {})) + + def test_from_callable_annotated_preserves_metadata_during_resolution(self): + annotation = typing.Annotated[typing.List['int'], 'metadata'] + resolved = decorators._resolve_annotation(annotation, globals(), locals()) + self.assertEqual(resolved, typing.Annotated[typing.List[int], 'metadata']) def test_from_callable_partial(self): def fn(a: int) -> int: diff --git a/sdks/python/apache_beam/typehints/typed_pipeline_test.py b/sdks/python/apache_beam/typehints/typed_pipeline_test.py index c97a1977e9c3..b98cc751a373 100644 --- a/sdks/python/apache_beam/typehints/typed_pipeline_test.py +++ b/sdks/python/apache_beam/typehints/typed_pipeline_test.py @@ -399,7 +399,7 @@ def process(self, element: int, **side_inputs: typehints.Dict[str, str]) \ _ = [1, 2, 3] | beam.ParDo(my_do_fn, a=1) def test_typed_callable_string_literals(self): - def do_fn(element: 'int') -> 'typehints.List[str]': + def do_fn(element: 'int') -> 'typehints.List[typehints.List[str]]': return [[str(element)] * 2] result = [1, 2] | beam.ParDo(do_fn) @@ -439,7 +439,8 @@ def fn(element: float): def test_typed_dofn_string_literals(self): class MyDoFn(beam.DoFn): - def process(self, element: 'int') -> 'typehints.List[str]': + def process( + self, element: 'int') -> 'typehints.List[typehints.List[str]]': return [[str(element)] * 2] result = [1, 2] | beam.ParDo(MyDoFn()) From 570ddb30e0f4aa3cc0590f45142788d9deb5327f Mon Sep 17 00:00:00 2001 From: Gaurav598 Date: Sat, 17 Oct 2026 04:18:56 +0530 Subject: [PATCH 4/4] Mark intentionally unresolved forward references in tests --- sdks/python/apache_beam/typehints/decorators_test.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/typehints/decorators_test.py b/sdks/python/apache_beam/typehints/decorators_test.py index bcff8673479d..3e9c3f182709 100644 --- a/sdks/python/apache_beam/typehints/decorators_test.py +++ b/sdks/python/apache_beam/typehints/decorators_test.py @@ -303,7 +303,7 @@ def process(self, value: 'int') -> 'typehints.List[str]': self.assertEqual(th.output_types, ((List[str], ), {})) def test_from_callable_unresolved_string_annotations_fall_back_to_any(self): - def fn(value: 'MissingType') -> 'int': + def fn(value: 'MissingType') -> 'int': # noqa: F821 return 1 th = decorators.IOTypeHints.from_callable(fn) @@ -343,7 +343,7 @@ def fn( self.assertEqual(th.output_types, ((List[str], ), {})) def test_from_callable_unresolved_nested_reference_falls_back_to_any(self): - def fn(values: typing.List['MissingType']) -> 'int': + def fn(values: typing.List['MissingType']) -> 'int': # noqa: F821 return len(values) th = decorators.IOTypeHints.from_callable(fn)