superset/utils/public_interfaces.py

# 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 base64 import b85encode
from hashlib import md5
from inspect import (
    getmembers,
    getsourcefile,
    getsourcelines,
    isclass,
    isfunction,
    isroutine,
    signature,
)
from textwrap import indent
from typing import Any, Callable


def compute_hash(obj: Callable[..., Any]) -> str:
    if isfunction(obj):
        return compute_func_hash(obj)

    if isclass(obj):
        return compute_class_hash(obj)

    raise Exception(f"Invalid object: {obj}")


def compute_func_hash(function: Callable[..., Any]) -> str:
    hashed = md5()
    hashed.update(str(signature(function)).encode())
    return b85encode(hashed.digest()).decode("utf-8")


def compute_class_hash(class_: Callable[..., Any]) -> str:
    hashed = md5()
    public_methods = sorted(
        [
            (name, method)
            for name, method in getmembers(class_, predicate=isroutine)
            if not name.startswith("_") or name == "__init__"
        ]
    )
    for name, method in public_methods:
        hashed.update(name.encode())
        hashed.update(str(signature(method)).encode())
    return b85encode(hashed.digest()).decode("utf-8")


def get_warning_message(obj: Callable[..., Any], expected_hash: str) -> str:
    sourcefile = getsourcefile(obj)
    sourcelines = getsourcelines(obj)
    code = indent("".join(sourcelines[0]), "    ")
    lineno = sourcelines[1]
    return (
        f"The object `{obj.__name__}` (in {sourcefile} "
        f"line {lineno}) has a public interface which has currently been "
        "modified. This MUST only be released in a new major version of "
        "Superset according to SIP-57. To remove this warning message "
        f"update the associated hash to '{expected_hash}'.\n\n{code}"
    )