diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py index 9bafce59f35ca..bba6bd9cc942b 100644 --- a/sdks/python/apache_beam/transforms/core.py +++ b/sdks/python/apache_beam/transforms/core.py @@ -21,6 +21,7 @@ import copy import inspect +import warnings import types from apache_beam import pvalue @@ -216,7 +217,7 @@ def is_process_bounded(self): # TODO(Sourabh): Remove after migration to NewDoFn -class DoFn(WithTypeHints, HasDisplayData): +class OldDoFn(WithTypeHints, HasDisplayData): """A function object used by a transform with custom processing. The ParDo transform is such a transform. The ParDo.expand() @@ -228,6 +229,10 @@ class DoFn(WithTypeHints, HasDisplayData): callable object using the CallableWrapperDoFn class. """ + def __init__(self): + warnings.warn('Use of OldDoFn is deprecated please use DoFn instead') + super(OldDoFn, self).__init__() + def default_label(self): return self.__class__.__name__ @@ -674,7 +679,7 @@ class ParDo(PTransformWithSideInputs): def __init__(self, fn_or_label, *args, **kwargs): super(ParDo, self).__init__(fn_or_label, *args, **kwargs) - if not isinstance(self.fn, (DoFn, NewDoFn)): + if not isinstance(self.fn, (OldDoFn, NewDoFn)): raise TypeError('ParDo must be called with a DoFn instance.') def default_type_hints(self): @@ -685,7 +690,7 @@ def infer_output_type(self, input_type): self.fn.infer_output_type(input_type)) def make_fn(self, fn): - if isinstance(fn, (DoFn, NewDoFn)): + if isinstance(fn, (OldDoFn, NewDoFn)): return fn return CallableWrapperDoFn(fn) diff --git a/sdks/python/apache_beam/typehints/typecheck.py b/sdks/python/apache_beam/typehints/typecheck.py index 7a10a5a2945b4..bc5583f7059ed 100644 --- a/sdks/python/apache_beam/typehints/typecheck.py +++ b/sdks/python/apache_beam/typehints/typecheck.py @@ -23,7 +23,7 @@ import types from apache_beam.pvalue import SideOutputValue -from apache_beam.transforms.core import DoFn +from apache_beam.transforms.core import OldDoFn from apache_beam.transforms.core import NewDoFn from apache_beam.transforms.window import WindowedValue from apache_beam.typehints import check_constraint @@ -35,7 +35,8 @@ from apache_beam.typehints.decorators import getcallargs_forhints -class TypeCheckWrapperDoFn(DoFn): +# TODO(Sourabh): Remove after migration to NewDoFn +class TypeCheckWrapperDoFn(OldDoFn): """A wrapper around a DoFn which performs type-checking of input and output. """ @@ -123,7 +124,8 @@ def _type_check(self, type_constraint, datum, is_input): raise TypeCheckError, error_msg, sys.exc_info()[2] -class OutputCheckWrapperDoFn(DoFn): +# TODO(Sourabh): Remove after migration to NewDoFn +class OutputCheckWrapperDoFn(OldDoFn): """A DoFn that verifies against common errors in the output type.""" def __init__(self, dofn, full_label):