osm_lcm/temporal/utils.py

#######################################################################################
# Copyright ETSI Contributors and Others.
#
# Licensed 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 temporalio.client import WorkflowFailureError


def get_root_cause(exception: WorkflowFailureError) -> BaseException:
    """Get the root cause of a WorkflowFailureError

    Temporal nests the cause of an exception at each layer as it bubbles up in
    the framework from activities and child workflows. This function will
    return the root cause of the exception.
    """
    cause = getattr(exception, "cause", None)
    while cause:
        exception = cause
        cause = getattr(exception, "cause", None)
    return exception