Source code for airflow.providers.google.common.deprecated

# 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.
from __future__ import annotations

import inspect
import re
from datetime import date, datetime
from typing import Any, Callable

from deprecated import deprecated as standard_deprecated
from deprecated.classic import ClassicAdapter


[docs]class AirflowDeprecationAdapter(ClassicAdapter): """ Build a detailed deprecation message based on the wrapped object type and other provided details. :param planned_removal_date: The date after which the deprecated object should be removed. The recommended date is six months ahead from today. The expected date format is `Month DD, YYYY`, for example: `August 22, 2024`. This parameter is required if the `planned_removal_release` parameter is not set. :param planned_removal_release: The package name and the version in which the deprecated object is expected to be removed. The expected format is `<package_name>==<package_version>`, for example `apache-airflow==2.10.0` or `apache-airflow-providers-google==10.22.0`. This parameter is required if the `planned_removal_date` parameter is not set. :param use_instead: Optional. Replacement of the deprecated object. :param reason: Optional. The detailed reason for the deprecated object. :param instructions: Optional. The detailed instructions for migrating from the deprecated object. :param category: Optional. The warning category to be used for the deprecation warning. """ def __init__( self, planned_removal_date: str | None = None, planned_removal_release: str | None = None, use_instead: str | None = None, reason: str | None = None, instructions: str | None = None, category: type[DeprecationWarning] = DeprecationWarning, **kwargs: Any, ): super().__init__(**kwargs) self.planned_removal_date: date | None = self._validate_date(planned_removal_date) self.planned_removal_release: str | None = self._validate_removal_release(planned_removal_release) self.use_instead: str | None = use_instead self.reason: str = reason or "" self.instructions: str | None = instructions self.category: type[DeprecationWarning] = category self._validate_fields()
[docs] def get_deprecated_msg(self, wrapped: Callable, instance: Any): """ Generate a deprecation message for wrapped callable. :param wrapped: Deprecated entity. :param instance: The instance to which the callable belongs. (not used) :return: A formatted deprecation message with all the details. """ entity_type = self.entity_type(entity=wrapped) entity_path = self.entity_path(entity=wrapped) sunset = self.sunset_message() replacement = self.replacement_message() msg = f"The {entity_type} `{entity_path}` is deprecated and will be removed {sunset}. {replacement}" if self.reason: msg += f" The reason is: {self.reason}" if self.instructions: msg += f" Instructions: {self.instructions}" return msg
@staticmethod def _validate_date(value: str | None) -> date | None: if value: try: return datetime.strptime(value, "%B %d, %Y").date() except ValueError as ex: error_message = ( f"Invalid date '{value}'. " f"The expected format is 'Month DD, YYYY', for example 'August 22, 2024'." ) raise ValueError(error_message) from ex return None @staticmethod def _validate_removal_release(value: str | None) -> str | None: if value: pattern = r"^apache-airflow(-providers-[a-zA-Z-]+)?==\d+\.\d+\.\d+.*$" if not bool(re.match(pattern, value)): raise ValueError( f"`{value}` must follow the format 'apache-airflow(-providers-<name>)==<X.Y.Z>'." ) return value def _validate_fields(self): msg = "Only one of two parameters must be set: `planned_removal_date` or 'planned_removal_release'." if self.planned_removal_release and self.planned_removal_date: raise ValueError(f"{msg} You specified both.") @staticmethod
[docs] def entity_type(entity: Callable) -> str: return "class" if inspect.isclass(entity) else "function (or method)"
@staticmethod
[docs] def entity_path(entity: Callable) -> str: module_name = getattr(entity, "__module__", "") qualified_name = getattr(entity, "__qualname__", "") full_path = f"{module_name}.{qualified_name}".strip(".") if module_name and full_path: return full_path return str(entity)
[docs] def sunset_message(self) -> str: if self.planned_removal_date: return f"after {self.planned_removal_date.strftime('%B %d, %Y')}" if self.planned_removal_release: return f"since version {self.planned_removal_release}" return "in the future"
[docs] def replacement_message(self): if self.use_instead: replacements = ", ".join(f"`{replacement}`" for replacement in self.use_instead.split(", ")) return f"Please use {replacements} instead." return "There is no replacement."
[docs]def deprecated( *args, planned_removal_date: str | None = None, planned_removal_release: str | None = None, use_instead: str | None = None, reason: str | None = None, instructions: str | None = None, adapter_cls: type[AirflowDeprecationAdapter] = AirflowDeprecationAdapter, **kwargs, ): """ Mark a class, method or a function deprecated. :param planned_removal_date: The date after which the deprecated object should be removed. The recommended date is six months ahead from today. The expected date format is `Month DD, YYYY`, for example: `August 22, 2024`. This parameter is required if the `planned_removal_release` parameter is not set. :param planned_removal_release: The package name and the version in which the deprecated object is expected to be removed. The expected format is `<package_name>==<package_version>`, for example `apache-airflow==2.10.0` or `apache-airflow-providers-google==10.22.0`. This parameter is required if the `planned_removal_date` parameter is not set. :param use_instead: Optional. Replacement of the deprecated object. :param reason: Optional. The detailed reason for the deprecated object. :param instructions: Optional. The detailed instructions for migrating from the deprecated object. :param adapter_cls: Optional. Adapter class that is used to get the deprecation message This should be a subclass of `AirflowDeprecationAdapter`. """ _kwargs = { **kwargs, "planned_removal_date": planned_removal_date, "planned_removal_release": planned_removal_release, "use_instead": use_instead, "reason": reason, "instructions": instructions, "adapter_cls": adapter_cls, } return standard_deprecated(*args, **_kwargs)

Was this entry helpful?