This is an automated email from the ASF dual-hosted git repository. dianfu pushed a commit to branch release-2.2 in repository https://gitbox.apache.org/repos/asf/flink.git
commit f84ec10c0d1bc9ea1acea4cb22b4a0a9986729e7 Author: Deepyaman Datta <[email protected]> AuthorDate: Wed Sep 9 23:52:21 2026 -0600 [FLINK-40529][python] Warn on use of deprecated APIs not at import time (#29061) --- flink-python/dev/integration_test.sh | 3 + flink-python/pyflink/table/table.py | 14 +- .../pyflink/util/api_stability_decorators.py | 78 +++- flink-python/pyflink/util/tests/__init__.py | 17 + .../util/tests/test_api_stability_decorators.py | 443 +++++++++++++++++++++ flink-python/pyproject.toml | 2 + flink-python/setup.py | 3 + 7 files changed, 543 insertions(+), 17 deletions(-) diff --git a/flink-python/dev/integration_test.sh b/flink-python/dev/integration_test.sh index c587a79f123..5689e85fd11 100755 --- a/flink-python/dev/integration_test.sh +++ b/flink-python/dev/integration_test.sh @@ -39,6 +39,9 @@ function test_all_modules() { # test table module test_module "table" + + # test util module + test_module "util" } # CURRENT_DIR is "flink/flink-python/dev/" diff --git a/flink-python/pyflink/table/table.py b/flink-python/pyflink/table/table.py index 285dd9d811c..c13d8209381 100644 --- a/flink-python/pyflink/table/table.py +++ b/flink-python/pyflink/table/table.py @@ -116,10 +116,11 @@ class Table(object): >>> tab.select(tab.a) """ - if name not in self.get_schema().get_field_names(): + column_names = self.get_resolved_schema().get_column_names() + if name not in column_names: raise AttributeError( "The current table has no column named '%s', available columns: [%s]" - % (name, ', '.join(self.get_schema().get_field_names()))) + % (name, ', '.join(column_names))) return col(name) def select(self, *fields: Expression) -> 'Table': @@ -943,27 +944,26 @@ class Table(object): .get(gateway.jvm.org.apache.flink.python.PythonOptions.MAX_ARROW_BATCH_SIZE) batches_iterator = gateway.jvm.org.apache.flink.table.runtime.arrow.ArrowUtils\ .collectAsPandasDataFrame(self._j_table, max_arrow_batch_size) + schema = self.get_schema() if batches_iterator.hasNext(): import pytz timezone = pytz.timezone( self._j_table.getTableEnvironment().getConfig().getLocalTimeZone().getId()) serializer = ArrowSerializer( - create_arrow_schema(self.get_schema().get_field_names(), - self.get_schema().get_field_data_types()), - self.get_schema().to_row_data_type(), + create_arrow_schema(schema.get_field_names(), schema.get_field_data_types()), + schema.to_row_data_type(), timezone) import pyarrow as pa table = pa.Table.from_batches(serializer.load_from_iterator(batches_iterator)) pdf = table.to_pandas() - schema = self.get_schema() for field_name in schema.get_field_names(): pdf[field_name] = tz_convert_from_internal( pdf[field_name], schema.get_field_data_type(field_name), timezone) return pdf else: import pandas as pd - return pd.DataFrame.from_records([], columns=self.get_schema().get_field_names()) + return pd.DataFrame.from_records([], columns=schema.get_field_names()) @Deprecated(since="2.1.0", detail="Use :func:`Table.get_resolved_schema` instead.") def get_schema(self) -> TableSchema: diff --git a/flink-python/pyflink/util/api_stability_decorators.py b/flink-python/pyflink/util/api_stability_decorators.py index abfa508cd4a..ba5b568b43f 100644 --- a/flink-python/pyflink/util/api_stability_decorators.py +++ b/flink-python/pyflink/util/api_stability_decorators.py @@ -16,11 +16,12 @@ # limitations under the License. ################################################################################ +import functools from inspect import getmembers, isfunction, isclass -from typing import TypeVar, Callable, Any, Union, Type, Optional +from typing import TypeVar, Callable, Any, Union, Type, Optional, cast from abc import ABCMeta, abstractmethod import warnings -from typing_extensions import override +from typing_extensions import deprecated, override from textwrap import dedent, indent __all__ = ["Deprecated", "Experimental", "Internal", "PublicEvolving", "Public"] @@ -84,7 +85,12 @@ class BaseAPIStabilityDecorator(metaclass=ABCMeta): stability_decorators = getattr(func_or_cls, '__stability_decorators') stability_decorators.add(self.__class__) else: - setattr(func_or_cls, '__stability_decorators', {self.__class__}) + # A property rejects attribute assignment, so it goes unrecorded rather + # than failing the import. + try: + setattr(func_or_cls, '__stability_decorators', {self.__class__}) + except (AttributeError, TypeError): + pass if isclass(func_or_cls): for name, method in getmembers( @@ -126,20 +132,72 @@ class Deprecated(BaseAPIStabilityDecorator): self.detail = detail def get_directive(self, func_or_cls: T) -> str: - return f".. deprecated:: {self.since}\n{indent(dedent(self.detail), ' ')}" + directive = f".. deprecated:: {self.since}" + if self.detail is not None: + directive = f"{directive}\n{indent(dedent(self.detail), ' ')}" + return directive - @override - def __call__(self, func_or_cls: T) -> T: + def _get_message(self, func_or_cls: T) -> str: """ - Emit a warning on the deprecation of the given function/class. Then call the base class - for docstring modification. + Returns the warning message emitted when the deprecated API element is used. """ msg = f"{func_or_cls.__qualname__} has been deprecated since version {self.since}." if self.detail is not None: msg = f"{msg} {self.detail}" + return msg + + @override + def __call__(self, func_or_cls: T) -> T: + """ + Arranges for a :class:`DeprecationWarning` to be emitted when the decorated API + element is *used*, and calls the base class for docstring modification. + + The warning cannot be emitted here: this runs while the module defining the API is + being imported, so it would warn every user who imports PyFlink and never the ones + who use the deprecated API. + """ + # typing_extensions.deprecated rejects a classmethod, and turns a staticmethod into + # a plain function that breaks when called on an instance. + if isinstance(func_or_cls, (staticmethod, classmethod)): + return cast(T, type(func_or_cls)(self(func_or_cls.__func__))) + + func_or_cls = super().__call__(func_or_cls) - warnings.warn(msg, category=DeprecationWarning, stacklevel=2) - return super().__call__(func_or_cls) + if isclass(func_or_cls): + self._deprecate_class(func_or_cls) + elif isfunction(func_or_cls): + return cast(T, deprecated(self._get_message(func_or_cls))(func_or_cls)) + + # A property is neither, and typing_extensions.deprecated rejects it, so it keeps + # the docstring directive alone. + return func_or_cls + + def _deprecate_class(self, cls: Type[Any]) -> None: + """ + Wraps the __init__ of the given class so that instantiating it warns, leaving the + class object itself in place so that isinstance checks and subclassing keep working. + + typing_extensions.deprecated is not used here: following PEP 702 it also warns when + a deprecated class is subclassed, and PyFlink subclasses its own deprecated classes + at module level, so that would warn on import. + """ + msg = self._get_message(cls) + original_init = cls.__init__ + + @functools.wraps(original_init) + def __init__(self: Any, *args: Any, **kwargs: Any) -> None: + # As in PEP 702, only the deprecated class itself warns, never a subclass, + # which may not be deprecated and would otherwise warn twice. + if type(self) is cls: + warnings.warn(msg, category=DeprecationWarning, stacklevel=2) + if original_init is object.__init__ and (args or kwargs) \ + and type(self).__new__ is object.__new__: + # object.__new__ raises this itself for a class that defines neither + # __new__ nor __init__; the __init__ installed here would mask it. + raise TypeError(f"{type(self).__name__}() takes no arguments") + original_init(self, *args, **kwargs) + + cls.__init__ = __init__ # type: ignore[misc] class Experimental(BaseAPIStabilityDecorator): diff --git a/flink-python/pyflink/util/tests/__init__.py b/flink-python/pyflink/util/tests/__init__.py new file mode 100644 index 00000000000..65b48d4d79b --- /dev/null +++ b/flink-python/pyflink/util/tests/__init__.py @@ -0,0 +1,17 @@ +################################################################################ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +################################################################################ diff --git a/flink-python/pyflink/util/tests/test_api_stability_decorators.py b/flink-python/pyflink/util/tests/test_api_stability_decorators.py new file mode 100644 index 00000000000..23cf1669d06 --- /dev/null +++ b/flink-python/pyflink/util/tests/test_api_stability_decorators.py @@ -0,0 +1,443 @@ +################################################################################ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +################################################################################ +import abc +import enum +import inspect +import os +import subprocess +import sys +import textwrap +import unittest +import warnings + +from pyflink.util.api_stability_decorators import ( + Deprecated, + Experimental, + Internal, + Public, + PublicEvolving, +) + + +class DeprecatedTests(unittest.TestCase): + """ + Tests for the :class:`Deprecated` decorator, which must warn when a deprecated API is + used, and not when it is defined. + + Blocks that must not warn turn warnings into errors, so that one fails where it is + raised rather than in a comparison afterwards. + """ + + def test_decoration_does_not_warn(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + + @Deprecated(since="1.0.0", detail="Use :func:`new_func` instead.") + def func(): + pass + + @Deprecated(since="1.0.0") + class Cls: + def __init__(self): + pass + + def test_importing_pyflink_table_does_not_warn(self): + # A fresh interpreter is the only way to observe an import: pyflink.table is + # already in sys.modules here, so importing it again is a no-op. Only this + # decorator's own warnings are inspected, so third-party noise cannot fail it. + script = textwrap.dedent( + """ + import warnings + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + import pyflink.table + import pyflink.table.descriptors + + print([str(warning.message) for warning in caught + if "has been deprecated since version" in str(warning.message)]) + """ + ) + result = subprocess.run( + [sys.executable, "-c", script], capture_output=True, text=True + ) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertEqual("[]", result.stdout.strip()) + + def test_function_warns_when_called(self): + @Deprecated(since="1.0.0", detail="Use :func:`new_func` instead.") + def func(a, b=2): + return a + b + + with self.assertWarns(DeprecationWarning) as caught: + self.assertEqual(3, func(1)) + + self.assertEqual( + f"{func.__qualname__} has been deprecated since version 1.0.0. " + f"Use :func:`new_func` instead.", + str(caught.warning), + ) + + def test_function_without_detail_warns_when_called(self): + @Deprecated(since="1.0.0") + def func(): + pass + + with self.assertWarns(DeprecationWarning) as caught: + func() + + self.assertEqual( + f"{func.__qualname__} has been deprecated since version 1.0.0.", + str(caught.warning), + ) + + def test_function_warns_once_per_call(self): + @Deprecated(since="1.0.0") + def func(): + pass + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + func() + + self.assertEqual(1, len(caught)) + + def test_function_wrapper_preserves_metadata(self): + @Deprecated(since="1.0.0") + def func(): + """Some documentation.""" + + self.assertEqual("func", func.__name__) + self.assertEqual( + "DeprecatedTests.test_function_wrapper_preserves_metadata.<locals>.func", + func.__qualname__, + ) + self.assertIn("Some documentation.", func.__doc__) + + def test_class_warns_when_instantiated(self): + @Deprecated(since="1.0.0", detail="Use :class:`NewClass` instead.") + class Cls: + def __init__(self, x): + self.x = x + + with self.assertWarns(DeprecationWarning) as caught: + instance = Cls(1) + + self.assertEqual(1, instance.x) + self.assertEqual( + f"{Cls.__qualname__} has been deprecated since version 1.0.0. " + f"Use :class:`NewClass` instead.", + str(caught.warning), + ) + + def test_class_warns_once_per_instantiation(self): + @Deprecated(since="1.0.0") + class Cls: + def __init__(self): + pass + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + Cls() + + self.assertEqual(1, len(caught)) + + def test_class_without_own_init_warns_when_instantiated(self): + @Deprecated(since="1.0.0") + class Cls: + pass + + with self.assertWarns(DeprecationWarning) as caught: + Cls() + + self.assertEqual( + f"{Cls.__qualname__} has been deprecated since version 1.0.0.", + str(caught.warning), + ) + + # Warning about the class must not swallow the error an argument would have raised. + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + with self.assertRaises(TypeError): + Cls(1) + + def test_class_is_returned_unchanged(self): + @Deprecated(since="1.0.0") + class Cls: + def __init__(self): + self.x = 1 + + class Subclass(Cls): + pass + + with warnings.catch_warnings(): + warnings.simplefilter("error") + instance = Subclass() + + self.assertIsInstance(instance, Cls) + self.assertTrue(issubclass(Subclass, Cls)) + self.assertEqual("Cls", Cls.__name__) + + def test_subclass_of_deprecated_class_does_not_warn(self): + # As in PEP 702: deprecating a class says nothing about its subclasses, which + # are the ones users are typically pointed at. + @Deprecated(since="1.0.0") + class Cls: + def __init__(self): + pass + + class Subclass(Cls): + pass + + with warnings.catch_warnings(): + warnings.simplefilter("error") + Subclass() + + def test_defining_a_subclass_does_not_warn(self): + # PEP 702 also warns when a deprecated class is subclassed. PyFlink subclasses + # its own deprecated classes at module level, so that would warn on import. + @Deprecated(since="1.0.0") + class Cls: + def __init__(self): + pass + + with warnings.catch_warnings(): + warnings.simplefilter("error") + + class Subclass(Cls): + pass + + def test_deprecated_subclass_inheriting_init_warns_once(self): + @Deprecated(since="1.0.0") + class Cls: + def __init__(self): + pass + + @Deprecated(since="2.0.0") + class Subclass(Cls): + pass + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + Subclass() + + self.assertEqual( + [f"{Subclass.__qualname__} has been deprecated since version 2.0.0."], + [str(warning.message) for warning in caught], + ) + + def test_function_warning_points_at_the_caller(self): + @Deprecated(since="1.0.0") + def func(): + pass + + with self.assertWarns(DeprecationWarning) as caught: + lineno = inspect.currentframe().f_lineno + 1 + func() + + self.assertEqual(os.path.abspath(__file__), os.path.abspath(caught.filename)) + self.assertEqual(lineno, caught.lineno) + + def test_class_warning_points_at_the_caller(self): + @Deprecated(since="1.0.0") + class Cls: + def __init__(self): + pass + + with self.assertWarns(DeprecationWarning) as caught: + lineno = inspect.currentframe().f_lineno + 1 + Cls() + + self.assertEqual(os.path.abspath(__file__), os.path.abspath(caught.filename)) + self.assertEqual(lineno, caught.lineno) + + def test_docstring_directives_are_still_applied(self): + @Deprecated(since="1.0.0", detail="Use :func:`new_func` instead.") + def func(): + """Function documentation.""" + + @Deprecated(since="1.0.0") + class Cls: + """Class documentation.""" + + def method(self): + """Method documentation.""" + + self.assertEqual( + "Function documentation.\n.. deprecated:: 1.0.0\n Use :func:`new_func` instead.", + func.__doc__, + ) + self.assertEqual("Class documentation.\n.. deprecated:: 1.0.0", Cls.__doc__) + self.assertEqual("Method documentation.\n.. deprecated:: 1.0.0", Cls.method.__doc__) + + def test_stability_decorators_attribute_is_still_populated(self): + @Deprecated(since="1.0.0") + def func(): + pass + + @Deprecated(since="1.0.0") + @PublicEvolving() + class Cls: + def __init__(self): + pass + + self.assertEqual({Deprecated}, getattr(func, "__stability_decorators")) + self.assertEqual({Deprecated, PublicEvolving}, getattr(Cls, "__stability_decorators")) + + def test_static_and_class_methods(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + + class Cls: + @Deprecated(since="1.0.0") + @staticmethod + def static_method(x): + """Static method documentation.""" + return x + + @Deprecated(since="1.0.0") + @classmethod + def class_method(cls, x): + return x + + # The descriptors must survive: a plain function wrapper around a staticmethod + # breaks when called on an instance. + self.assertIsInstance(Cls.__dict__["static_method"], staticmethod) + self.assertIsInstance(Cls.__dict__["class_method"], classmethod) + self.assertIn(".. deprecated:: 1.0.0", Cls.static_method.__doc__) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + self.assertEqual(1, Cls.static_method(1)) + self.assertEqual(2, Cls.class_method(2)) + self.assertEqual(3, Cls().static_method(3)) + self.assertEqual(4, Cls().class_method(4)) + + expected = [ + f"{Cls.__qualname__}.static_method has been deprecated since version 1.0.0.", + f"{Cls.__qualname__}.class_method has been deprecated since version 1.0.0.", + ] + # Once through the class, once through an instance. + self.assertEqual(expected * 2, [str(warning.message) for warning in caught]) + + def test_property(self): + # A property only gets the docstring directive, but must not raise. + with warnings.catch_warnings(): + warnings.simplefilter("error") + + class Cls: + @Deprecated(since="1.0.0") + @property + def value(self): + """Property documentation.""" + return 1 + + self.assertEqual(1, Cls().value) + + self.assertIn(".. deprecated:: 1.0.0", Cls.__dict__["value"].__doc__) + + def test_abstract_class(self): + with warnings.catch_warnings(): + warnings.simplefilter("error") + + @Deprecated(since="1.0.0") + class Abstract(abc.ABC): + """Abstract class documentation.""" + + @abc.abstractmethod + def method(self): + pass + + with self.assertRaises(TypeError): + Abstract() + + self.assertIn(".. deprecated:: 1.0.0", Abstract.__doc__) + + def test_enum_class(self): + # Members are created before the decorator runs, so they are unaffected; + # decorating an Enum must not raise and must leave lookup working. + with warnings.catch_warnings(): + warnings.simplefilter("error") + + @Deprecated(since="1.0.0") + class Colour(enum.Enum): + """Enum documentation.""" + + RED = 1 + + self.assertIs(Colour.RED, Colour(1)) + self.assertEqual(1, Colour.RED.value) + self.assertEqual([Colour.RED], list(Colour)) + + self.assertIn(".. deprecated:: 1.0.0", Colour.__doc__) + + +class OtherStabilityDecoratorTests(unittest.TestCase): + """ + Tests that the decorators other than :class:`Deprecated` document without warning. + """ + + def test_decorators_never_warn(self): + for decorator in (Experimental, Internal, Public, PublicEvolving): + with self.subTest(decorator=decorator.__name__): + with warnings.catch_warnings(): + warnings.simplefilter("error") + + @decorator() + def func(): + """Function documentation.""" + + @decorator() + class Cls: + """Class documentation.""" + + def method(self): + """Method documentation.""" + + func() + Cls().method() + + def test_decorated_elements_are_returned_unchanged(self): + for decorator in (Experimental, Internal, Public, PublicEvolving): + with self.subTest(decorator=decorator.__name__): + + def func(): + pass + + class Cls: + pass + + self.assertIs(func, decorator()(func)) + self.assertIs(Cls, decorator()(Cls)) + + def test_docstring_directives_and_attribute(self): + @Public() + class Cls: + """Class documentation.""" + + def method(self): + """Method documentation.""" + + self.assertIn("is marked as **public**", Cls.__doc__) + self.assertIn("is marked as **public**", Cls.method.__doc__) + self.assertEqual({Public}, getattr(Cls, "__stability_decorators")) + + +if __name__ == "__main__": + unittest.main() diff --git a/flink-python/pyproject.toml b/flink-python/pyproject.toml index d96141ed0e5..9b83c8f7aff 100644 --- a/flink-python/pyproject.toml +++ b/flink-python/pyproject.toml @@ -50,6 +50,8 @@ dev = [ "protobuf~=4.25", "pytest~=8.0", "ruamel.yaml>=0.18.4", + "typing-extensions>=4.5.0; python_version < '3.12'", + "typing-extensions>=4.7.0; python_version >= '3.12'", ] tox = [ "tox==3.14.0" diff --git a/flink-python/setup.py b/flink-python/setup.py index 89473d441a7..452a8760610 100644 --- a/flink-python/setup.py +++ b/flink-python/setup.py @@ -328,6 +328,9 @@ try: 'pemja>=0.5.8,<0.5.9;platform_system != "Windows"', 'httplib2>=0.19.0', 'ruamel.yaml>=0.18.4', + # deprecated() landed in 4.5.0; 4.7.0 declares 3.12 support. + 'typing-extensions>=4.5.0;python_version < "3.12"', + 'typing-extensions>=4.7.0;python_version >= "3.12"', apache_flink_libraries_dependency] setup(
