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)