HierarchyService

HierarchyService(rest)

Bases: ObjectService

Service to handle Object Updates for TM1 Hierarchies

Source code in TM1py/Services/HierarchyService.py

def __init__(self, rest: RestService):
    super().__init__(rest)
    self.subsets = SubsetService(rest)
    self.elements = ElementService(rest)

EDGES_WORKAROUND_VERSIONS = ('11.0.002', '11.0.003', '11.1.000') class-attribute instance-attribute

HIERARCHY_SORT_ORDER_ARGUMENTS = CaseAndSpaceInsensitiveDict({'CompSortType': CaseAndSpaceInsensitiveSet(['ByInput', 'ByName']), 'CompSortSense': CaseAndSpaceInsensitiveSet(['Ascending', 'Descending']), 'ElSortType': CaseAndSpaceInsensitiveSet(['ByInput', 'ByName', 'ByLevel', 'ByHierarchy']), 'ElSortSense': CaseAndSpaceInsensitiveSet(['Ascending', 'Descending'])}) class-attribute instance-attribute

elements = ElementService(rest) instance-attribute

subsets = SubsetService(rest) instance-attribute

add_edges(dimension_name, hierarchy_name=None, edges=None, **kwargs)

Add Edges to hierarchy. Fails if one edge already exists.

Parameters:

Name	Type	Description	Default
dimension_name	str		required
hierarchy_name	str		None
edges	Dict[Tuple[str, str], int]		None

Returns:

Type	Description
Response

Source code in TM1py/Services/HierarchyService.py

def add_edges(
    self, dimension_name: str, hierarchy_name: str = None, edges: Dict[Tuple[str, str], int] = None, **kwargs
) -> Response:
    """Add Edges to hierarchy. Fails if one edge already exists.

    :param dimension_name:
    :param hierarchy_name:
    :param edges:
    :return:
    """
    return self.elements.add_edges(dimension_name, hierarchy_name, edges, **kwargs)

add_element_attributes(dimension_name, hierarchy_name, element_attributes, **kwargs)

Add element attributes to hierarchy. Fails if one element attribute already exists.

Parameters:

Name	Type	Description	Default
dimension_name	str		required
hierarchy_name	str		required
element_attributes	List[ElementAttribute]		required

Returns:

Type	Description

Source code in TM1py/Services/HierarchyService.py

def add_element_attributes(
    self, dimension_name: str, hierarchy_name: str, element_attributes: List[ElementAttribute], **kwargs
):
    """Add element attributes to hierarchy. Fails if one element attribute already exists.

    :param dimension_name:
    :param hierarchy_name:
    :param element_attributes:
    :return:
    """
    return self.elements.add_element_attributes(dimension_name, hierarchy_name, element_attributes, **kwargs)

add_elements(dimension_name, hierarchy_name, elements, **kwargs)

Add elements to hierarchy. Fails if one element already exists.

Parameters:

Name	Type	Description	Default
dimension_name	str		required
hierarchy_name	str		required
elements	List[Element]		required

Returns:

Type	Description

Source code in TM1py/Services/HierarchyService.py

def add_elements(self, dimension_name: str, hierarchy_name: str, elements: List[Element], **kwargs):
    """Add elements to hierarchy. Fails if one element already exists.

    :param dimension_name:
    :param hierarchy_name:
    :param elements:
    :return:
    """
    return self.elements.add_elements(dimension_name, hierarchy_name, elements, **kwargs)

create(hierarchy, **kwargs)

Create a hierarchy in an existing dimension

Parameters:

Name	Type	Description	Default
hierarchy	Hierarchy		required

Returns:

Type	Description

Source code in TM1py/Services/HierarchyService.py

def create(self, hierarchy: Hierarchy, **kwargs):
    """Create a hierarchy in an existing dimension

    :param hierarchy:
    :return:
    """
    url = format_url("/Dimensions('{}')/Hierarchies", hierarchy.dimension_name)
    response = self._rest.POST(url, hierarchy.body, **kwargs)

    self.update_element_attributes(hierarchy, **kwargs)

    return response

delete(dimension_name, hierarchy_name, **kwargs)

Source code in TM1py/Services/HierarchyService.py

def delete(self, dimension_name: str, hierarchy_name: str, **kwargs) -> Response:
    url = format_url("/Dimensions('{}')/Hierarchies('{}')", dimension_name, hierarchy_name)
    return self._rest.DELETE(url, **kwargs)

exists(dimension_name, hierarchy_name, **kwargs)

Parameters:

Name	Type	Description	Default
dimension_name	str		required
hierarchy_name	str		required

Returns:

Type	Description
bool

Source code in TM1py/Services/HierarchyService.py

def exists(self, dimension_name: str, hierarchy_name: str, **kwargs) -> bool:
    """

    :param dimension_name:
    :param hierarchy_name:
    :return:
    """
    url = format_url("/Dimensions('{}')/Hierarchies?$select=Name", dimension_name)

    try:
        response = self._rest.GET(url, **kwargs)
    except TM1pyRestException as e:
        if e.status_code == 404:
            return False
        raise e

    existing_hierarchies = CaseAndSpaceInsensitiveSet([hierarchy["Name"] for hierarchy in response.json()["value"]])
    return hierarchy_name in existing_hierarchies

get(dimension_name, hierarchy_name, **kwargs)

get hierarchy

Parameters:

Name	Type	Description	Default
dimension_name	str	name of the dimension	required
hierarchy_name	str	name of the hierarchy	required

Returns:

Type	Description
Hierarchy

Source code in TM1py/Services/HierarchyService.py

def get(self, dimension_name: str, hierarchy_name: str, **kwargs) -> Hierarchy:
    """get hierarchy

    :param dimension_name: name of the dimension
    :param hierarchy_name: name of the hierarchy
    :return:
    """
    url = format_url(
        "/Dimensions('{}')/Hierarchies('{}')?$expand=Edges,Elements,ElementAttributes,Subsets,DefaultMember",
        dimension_name,
        hierarchy_name,
    )
    response = self._rest.GET(url, **kwargs)
    return Hierarchy.from_dict(response.json())

get_all_names(dimension_name, **kwargs)

get all names of existing Hierarchies in a dimension

Parameters:

Name	Type	Description	Default
dimension_name	str		required

Returns:

Type	Description
List[str]

Source code in TM1py/Services/HierarchyService.py

def get_all_names(self, dimension_name: str, **kwargs) -> List[str]:
    """get all names of existing Hierarchies in a dimension

    :param dimension_name:
    :return:
    """
    url = format_url("/Dimensions('{}')/Hierarchies?$select=Name", dimension_name)
    response = self._rest.GET(url, **kwargs)
    return [hierarchy["Name"] for hierarchy in response.json()["value"]]

get_cell_service()

Source code in TM1py/Services/HierarchyService.py

def get_cell_service(self):
    from TM1py import CellService

    return CellService(self._rest)

get_default_member(dimension_name, hierarchy_name=None, **kwargs)

Get the defined default_member for a Hierarchy. Will return the element with index 1, if default member is not specified explicitly in }HierarchyProperty Cube

Parameters:

Name	Type	Description	Default
dimension_name	str		required
hierarchy_name	str		None

Returns:

Type	Description
Optional[str]	String, name of Member

Source code in TM1py/Services/HierarchyService.py

def get_default_member(self, dimension_name: str, hierarchy_name: str = None, **kwargs) -> Optional[str]:
    """Get the defined default_member for a Hierarchy.
    Will return the element with index 1, if default member is not specified explicitly in }HierarchyProperty Cube

    :param dimension_name:
    :param hierarchy_name:
    :return: String, name of Member
    """
    url = format_url(
        "/Dimensions('{dimension}')/Hierarchies('{hierarchy}')/DefaultMember",
        dimension=dimension_name,
        hierarchy=hierarchy_name if hierarchy_name else dimension_name,
    )
    response = self._rest.GET(url=url, **kwargs)

    if not response.text:
        return None
    return response.json()["Name"]

get_dimension_service()

Source code in TM1py/Services/HierarchyService.py

def get_dimension_service(self):
    from TM1py import DimensionService

    return DimensionService(self._rest)

get_hierarchy_summary(dimension_name, hierarchy_name, **kwargs)

Source code in TM1py/Services/HierarchyService.py

def get_hierarchy_summary(self, dimension_name: str, hierarchy_name: str, **kwargs) -> Dict[str, int]:
    hierarchy_properties = ("Elements", "Edges", "ElementAttributes", "Members", "Levels")
    url = format_url(
        "/Dimensions('{}')/Hierarchies('{}')?$expand=Edges/$count,Elements/$count,"
        "ElementAttributes/$count,Members/$count,Levels/$count&$select=Cardinality",
        dimension_name,
        hierarchy_name,
    )
    hierary_summary_raw = self._rest.GET(url, **kwargs).json()

    return {
        hierarchy_property: hierary_summary_raw[hierarchy_property + "@odata.count"]
        for hierarchy_property in hierarchy_properties
    }

is_balanced(dimension_name, hierarchy_name, **kwargs)

Check if hierarchy is balanced

Parameters:

Name	Type	Description	Default
dimension_name	str		required
hierarchy_name	str		required

Returns:

Type	Description

Source code in TM1py/Services/HierarchyService.py

def is_balanced(self, dimension_name: str, hierarchy_name: str, **kwargs):
    """Check if hierarchy is balanced

    :param dimension_name:
    :param hierarchy_name:
    :return:
    """
    url = format_url("/Dimensions('{}')/Hierarchies('{}')/Structure/$value", dimension_name, hierarchy_name)
    structure = int(self._rest.GET(url, **kwargs).text)
    # 0 = balanced, 2 = unbalanced
    if structure == 0:
        return True
    elif structure == 2:
        return False
    else:
        raise RuntimeError(f"Unexpected return value from TM1 API request: {str(structure)}")

remove_all_edges(dimension_name, hierarchy_name=None, **kwargs)

Source code in TM1py/Services/HierarchyService.py

def remove_all_edges(self, dimension_name: str, hierarchy_name: str = None, **kwargs) -> Response:
    if not hierarchy_name:
        hierarchy_name = dimension_name
    url = format_url("/Dimensions('{}')/Hierarchies('{}')", dimension_name, hierarchy_name)
    body = {"Edges": []}
    return self._rest.PATCH(url=url, data=json.dumps(body), **kwargs)

remove_edges_under_consolidation(dimension_name, hierarchy_name, consolidation_element, **kwargs)

Parameters:

Name	Type	Description	Default
dimension_name	str	Name of the dimension	required
hierarchy_name	str	Name of the hierarchy	required
consolidation_element	str	Name of the Consolidated element	required

Returns:

Type	Description
List[Response]	response

Source code in TM1py/Services/HierarchyService.py

def remove_edges_under_consolidation(
    self, dimension_name: str, hierarchy_name: str, consolidation_element: str, **kwargs
) -> List[Response]:
    """
    :param dimension_name: Name of the dimension
    :param hierarchy_name: Name of the hierarchy
    :param consolidation_element: Name of the Consolidated element
    :return: response
    """
    hierarchy = self.get(dimension_name, hierarchy_name)
    from TM1py.Services import ElementService

    element_service = ElementService(self._rest)
    elements_under_consolidations = CaseAndSpaceInsensitiveSet(
        element_service.get_members_under_consolidation(dimension_name, hierarchy_name, consolidation_element)
    )
    elements_under_consolidations.add(consolidation_element)
    remove_edges = []
    for parent, component in hierarchy.edges:
        if parent in elements_under_consolidations and component in elements_under_consolidations:
            remove_edges.append((parent, component))
    hierarchy.remove_edges(remove_edges)
    return self.update(hierarchy, **kwargs)

update(hierarchy, keep_existing_attributes=False, **kwargs)

update a hierarchy. It's a two step process: 1. Update Hierarchy 2. Update Element-Attributes

Function caters for Bug with Edge Creation: https://www.ibm.com/developerworks/community/forums/html/topic?id=75f2b99e-6961-4c71-9364-1d5e1e083eff

Parameters:

Name	Type	Description	Default
hierarchy	Hierarchy	instance of TM1py.Hierarchy	required
keep_existing_attributes		True to make sure existing attributes are not removed	False

Returns:

Type	Description
List[Response]	list of responses

Source code in TM1py/Services/HierarchyService.py

def update(self, hierarchy: Hierarchy, keep_existing_attributes=False, **kwargs) -> List[Response]:
    """update a hierarchy. It's a two step process:
    1. Update Hierarchy
    2. Update Element-Attributes

    Function caters for Bug with Edge Creation:
    https://www.ibm.com/developerworks/community/forums/html/topic?id=75f2b99e-6961-4c71-9364-1d5e1e083eff

    :param hierarchy: instance of TM1py.Hierarchy
    :param keep_existing_attributes: True to make sure existing attributes are not removed
    :return: list of responses
    """
    # functions returns multiple responses
    responses = list()
    # 1. Update Hierarchy
    url = format_url("/Dimensions('{}')/Hierarchies('{}')", hierarchy.dimension_name, hierarchy.name)
    # Workaround EDGES: Handle Issue, that Edges cant be created in one batch with the Hierarchy in certain versions
    hierarchy_body = hierarchy.body_as_dict
    if self.version[0:8] in self.EDGES_WORKAROUND_VERSIONS:
        del hierarchy_body["Edges"]
    responses.append(self._rest.PATCH(url, json.dumps(hierarchy_body), **kwargs))

    # 2. Update Attributes
    responses.append(
        self.update_element_attributes(
            hierarchy=hierarchy, keep_existing_attributes=keep_existing_attributes, **kwargs
        )
    )

    # Workaround EDGES
    if self.version[0:8] in self.EDGES_WORKAROUND_VERSIONS:
        process_service = self._get_process_service()
        ti_function = "HierarchyElementComponentAdd('{}', '{}', '{}', '{}', {});"
        ti_statements = [
            ti_function.format(
                hierarchy.dimension_name, hierarchy.name, edge[0], edge[1], hierarchy.edges[(edge[0], edge[1])]
            )
            for edge in hierarchy.edges
        ]
        responses.append(process_service.execute_ti_code(lines_prolog=ti_statements, **kwargs))

    return responses

update_default_member(dimension_name, hierarchy_name=None, member_name='', **kwargs)

Update the default member of a hierarchy.

Parameters:

Name	Type	Description	Default
dimension_name	str		required
hierarchy_name	str		None
member_name	str		''

Returns:

Type	Description
Response

Source code in TM1py/Services/HierarchyService.py

def update_default_member(
    self, dimension_name: str, hierarchy_name: str = None, member_name: str = "", **kwargs
) -> Response:
    """Update the default member of a hierarchy.

    :param dimension_name:
    :param hierarchy_name:
    :param member_name:
    :return:
    """
    if verify_version(required_version="12", version=self.version):
        return self._update_default_member_via_api(dimension_name, hierarchy_name, member_name)
    else:
        return self._update_default_member_via_props_cube(dimension_name, hierarchy_name, member_name)

update_element_attributes(hierarchy, keep_existing_attributes=False, **kwargs)

Update the elementattributes of a hierarchy

Parameters:

Name	Type	Description	Default
hierarchy	Hierarchy	Instance of TM1py.Hierarchy	required
keep_existing_attributes		True to make sure existing attributes are not removed	False

Returns:

Type	Description

Source code in TM1py/Services/HierarchyService.py

def update_element_attributes(self, hierarchy: Hierarchy, keep_existing_attributes=False, **kwargs):
    """Update the elementattributes of a hierarchy

    :param hierarchy: Instance of TM1py.Hierarchy
    :param keep_existing_attributes: True to make sure existing attributes are not removed
    :return:
    """
    # get existing attributes first
    existing_element_attributes = self.elements.get_element_attributes(
        dimension_name=hierarchy.dimension_name, hierarchy_name=hierarchy.name, **kwargs
    )
    existing_element_attributes = CaseAndSpaceInsensitiveDict({ea.name: ea for ea in existing_element_attributes})

    attributes_to_create = list()
    attributes_to_delete = list()
    attributes_to_update = list()

    for element_attribute in hierarchy.element_attributes:
        if element_attribute.name not in existing_element_attributes:
            attributes_to_create.append(element_attribute)
            continue

        existing_element_attribute = existing_element_attributes[element_attribute.name]
        if not existing_element_attribute.attribute_type == element_attribute.attribute_type:
            attributes_to_update.append(element_attribute)
            continue

    if not keep_existing_attributes:
        for existing_element_attribute in existing_element_attributes:
            if existing_element_attribute not in CaseAndSpaceInsensitiveSet(
                [ea.name for ea in hierarchy.element_attributes]
            ):
                attributes_to_delete.append(existing_element_attribute)

    for element_attribute in attributes_to_create:
        self.elements.create_element_attribute(
            dimension_name=hierarchy.dimension_name,
            hierarchy_name=hierarchy.name,
            element_attribute=element_attribute,
            **kwargs,
        )

    for element_attribute in attributes_to_delete:
        self.elements.delete_element_attribute(
            dimension_name=hierarchy.dimension_name,
            hierarchy_name=hierarchy.name,
            element_attribute=element_attribute,
            **kwargs,
        )

    for element_attribute in attributes_to_update:
        self.elements.delete_element_attribute(
            dimension_name=hierarchy.dimension_name,
            hierarchy_name=hierarchy.name,
            element_attribute=element_attribute.name,
            **kwargs,
        )
        self.elements.create_element_attribute(
            dimension_name=hierarchy.dimension_name,
            hierarchy_name=hierarchy.name,
            element_attribute=element_attribute,
            **kwargs,
        )

update_or_create(hierarchy, **kwargs)

update if exists else create

Parameters:

Name	Type	Description	Default
hierarchy	Hierarchy		required

Returns:

Type	Description

Source code in TM1py/Services/HierarchyService.py

def update_or_create(self, hierarchy: Hierarchy, **kwargs):
    """update if exists else create

    :param hierarchy:
    :return:
    """
    if self.exists(dimension_name=hierarchy.dimension_name, hierarchy_name=hierarchy.name, **kwargs):
        self.update(hierarchy=hierarchy, **kwargs)
    else:
        self.create(hierarchy=hierarchy, **kwargs)

update_or_create_hierarchy_from_dataframe(dimension_name, hierarchy_name, df, element_column=None, verify_unique_elements=False, verify_edges=True, element_type_column='ElementType', unwind_all=False, unwind_consolidations=None, update_attribute_types=False, hierarchy_sort_order=None, delete_orphaned_consolidations=False, **kwargs)

Update or Create a hierarchy based on a dataframe, while never deleting existing elements.

Parameters:

Name	Type	Description	Default
dimension_name	str	Name of the dimension	required
hierarchy_name	str	Name of the hierarchy	required
df	DataFrame	pd.DataFrame the data frame. Example: | | Region | ElementType | Alias:a | Currency:s | population:n | level001 | level000 | level001_weight | level000_weight | |---:|:--------|:------------|:------------|:-----------|-------------:|:---------|:---------|----------------:|----------------:| | 0 | France | Numeric | Frankreich | EUR | 60000000 | Europe | World | 1 | 1 | | 1 | Belgium | Numeric | Schweiz | CHF | 9000000 | Europe | World | 1 | 1 | | 2 | Germany | Numeric | Deutschland | EUR | 84000000 | Europe | World | 1 | 1 | Names for the parent columns (level001, level000) are not configurable and level000 is the top node. All columns except for the element_column, element_type_colums and parent columns are attribute columns. On attribute columns, you specify the type as a suffix. If no type is provided string attributes are created	required
element_type_column	str	str The column name in the df which specifies which element is which type. If None, all will be considered N level.	'ElementType'
element_column	str	str The column name of the element ID. If None, assumes first column is the element ID.	None
verify_unique_elements	bool	Abort early if element names are not unique	False
verify_edges	bool	Abort early if edges contain a circular reference.	True
unwind_all	bool	bool Unwind hierarch before creating new edges	False
unwind_consolidations	Iterable	list Unwind a list of specific consolidations in the hierarchy before creating new edges, if unwind_all is true, this list is ignored	None
update_attribute_types	bool	bool If True, function will delete and recreate attributes when a type change is requested. By default, function will not delete attributes.	False
hierarchy_sort_order	Tuple[str, str, str, str]	Dict Pass a Tuple with 4 values for: CompSortType, CompSortSense, ElSortType, ElSortSense to control sort order as in IBM docs: https://www.ibm.com/docs/en/planning-analytics/2.0.0?topic=hmtf-hierarchysortorder-1	None
delete_orphaned_consolidations	bool	bool If True, function will delete c elements that will have no children and will have no parents post update. By default, function will not delete orphaned consolidations.	False

Returns:

Type	Description

Source code in TM1py/Services/HierarchyService.py

@require_pandas
@require_data_admin
@require_ops_admin
def update_or_create_hierarchy_from_dataframe(
    self,
    dimension_name: str,
    hierarchy_name: str,
    df: "pd.DataFrame",
    element_column: str = None,
    verify_unique_elements: bool = False,
    verify_edges: bool = True,
    element_type_column: str = "ElementType",
    unwind_all: bool = False,
    unwind_consolidations: Iterable = None,
    update_attribute_types: bool = False,
    hierarchy_sort_order: Tuple[str, str, str, str] = None,
    delete_orphaned_consolidations: bool = False,
    **kwargs,
):
    """Update or Create a hierarchy based on a dataframe, while never deleting existing elements.

    :param dimension_name:
        Name of the dimension
    :param hierarchy_name:
        Name of the hierarchy
    :param df: pd.DataFrame the data frame. Example:
        |    | Region  | ElementType | Alias:a     | Currency:s | population:n | level001 | level000 | level001_weight | level000_weight |
        |---:|:--------|:------------|:------------|:-----------|-------------:|:---------|:---------|----------------:|----------------:|
        |  0 | France  | Numeric     | Frankreich  | EUR        |     60000000 | Europe   | World    |               1 |               1 |
        |  1 | Belgium | Numeric     | Schweiz     | CHF        |      9000000 | Europe   | World    |               1 |               1 |
        |  2 | Germany | Numeric     | Deutschland | EUR        |     84000000 | Europe   | World    |               1 |               1 |

        Names for the parent columns (level001, level000) are not configurable and `level000` is the top node.
        All columns except for the element_column, element_type_colums and parent columns are attribute columns.
        On attribute columns, you specify the type as a suffix. If no type is provided string attributes are created

    :param element_type_column: str
        The column name in the df which specifies which element is which type.
        If None, all will be considered N level.
    :param element_column: str
        The column name of the element ID. If None, assumes first column is the element ID.
    :param verify_unique_elements:
        Abort early if element names are not unique
    :param verify_edges:
        Abort early if edges contain a circular reference.
    :param unwind_all: bool
        Unwind hierarch before creating new edges
    :param unwind_consolidations: list
        Unwind a list of specific consolidations in the hierarchy before creating new edges,
        if unwind_all is true, this list is ignored
    :param update_attribute_types: bool
        If True, function will delete and recreate attributes when a type change is requested.
        By default, function will not delete attributes.
    :param hierarchy_sort_order: Dict
        Pass a Tuple with 4 values for: `CompSortType`, `CompSortSense`, `ElSortType`, `ElSortSense` to control
        sort order as in IBM docs:
        https://www.ibm.com/docs/en/planning-analytics/2.0.0?topic=hmtf-hierarchysortorder-1
    :param delete_orphaned_consolidations: bool
        If True, function will delete c elements that will have no children and will have no parents post update.
        By default, function will not delete orphaned consolidations.

    :return:

    """
    if hierarchy_sort_order:
        self._validate_hierarchy_sort_order_arguments(hierarchy_sort_order)

    df = df.copy()

    # element ID is in first column if not specified.
    element_column = df.columns[0] if not element_column else element_column
    df[element_column] = df[element_column].astype(str)

    # assume all Numeric if no type is provided
    if element_type_column not in df.columns:
        df[element_type_column] = "Numeric"

    # verify uniqueness of element names
    if verify_unique_elements:
        unique_element_names = len(set(df[element_column].str.lower().str.replace(" ", "")))
        if df.shape[0] != unique_element_names:
            raise ValueError("There must be no duplicates in the element column")

    # verify alias uniqueness
    alias_columns = tuple([col for col in df.columns if col.lower().endswith((":a", ":alias"))])
    if len(alias_columns) > 0:
        self._validate_alias_uniqueness(df=df[[element_column, *alias_columns]])

    # backward compatibility for unwind, the value for unwind would be assinged to unwind_all. expected type is bool
    if "unwind" in kwargs:
        unwind_all = kwargs["unwind"]

    if unwind_consolidations:
        if isinstance(unwind_consolidations, str) or not isinstance(unwind_consolidations, Iterable):
            raise ValueError(
                f"value for 'unwind_consolidations' must be an iterable (e.g., list), "
                f"but received: '{unwind_consolidations}' of type {type(unwind_consolidations).__name__}"
            )

    # identify and sort level columns
    level_columns = []
    level_weight_columns = []
    # sort to assure right order of levels (e.g. Level003 -> level002 -> LEVEL001)
    sorted_level_columns = sorted(
        [col for col in df.columns if any(char.isdigit() for char in col)],  # Filter columns with digits
        key=lambda x: int("".join(filter(str.isdigit, x))),  # Sort based on numeric part
        reverse=True,  # Descending order
    )
    for column in sorted_level_columns:
        if column.lower().startswith("level") and column[5:8].isdigit():
            if len(column) == 8:  # "LevelXXX"
                level_columns.append(column)
            elif len(column) == 15 and column.lower().endswith("_weight"):  # "LevelXXX_weight"
                level_weight_columns.append(column)

    # case: no level weight columns. All weights are 1
    if len(level_weight_columns) == 0:
        for level_column in level_columns:
            level_weight_column = level_column + "_weight"
            level_weight_columns.append(level_weight_column)
            df[level_weight_column] = 1

    if not len(level_columns) == len(level_weight_columns):
        raise ValueError("Number of level columns must be equal to number of level weight columns")

    if verify_edges:
        self._validate_edges(df=df[[element_column, *level_columns]])

    hierarchy_exists = self.exists(dimension_name, hierarchy_name)

    if not hierarchy_exists:
        existing_element_identifiers = CaseAndSpaceInsensitiveSet()
    else:
        existing_element_identifiers = self.elements.get_all_element_identifiers(
            dimension_name=dimension_name, hierarchy_name=hierarchy_name
        )

    if not hierarchy_exists:
        hierarchy = Hierarchy(name=hierarchy_name, dimension_name=dimension_name)
        dimension_service = self.get_dimension_service()
        if not dimension_service.exists(dimension_name):
            dimension = Dimension(name=dimension_name, hierarchies=[hierarchy])
            dimension_service.create(dimension)
        else:
            hierarchy = Hierarchy(name=hierarchy_name, dimension_name=dimension_name)
            self.create(hierarchy)

    # determine new elements based on Element Name column
    new_elements = CaseAndSpaceInsensitiveDict(
        {
            element_name: Element.Types(element_type)
            for element_name, element_type in df.loc[
                ~df[element_column]
                .str.lower()
                .str.replace(" ", "")
                .isin(existing_element_identifiers._store.keys()),
                (element_column, element_type_column),
            ].itertuples(index=False)
        }
    )

    # determine new consolidations based on level columns
    for element_name in df[[*level_columns]].stack().unique():
        if not element_name:
            continue
        if element_name in existing_element_identifiers:
            continue
        if element_name in new_elements and new_elements[element_name] != Element.Types.CONSOLIDATED:
            raise ValueError(f"Inconsistent Type for element: '{element_name}' in hierarchy '{hierarchy_name}'")
        new_elements[element_name] = Element.Types.CONSOLIDATED

    if new_elements:
        # add these elements to hierarchy in tm1
        self.elements.add_elements(
            dimension_name=dimension_name,
            hierarchy_name=hierarchy_name,
            elements=(Element(element_name, element_type) for element_name, element_type in new_elements.items()),
        )

    # define the attribute columns in df. Applies to all elements in df, not only new ones.
    attribute_columns = df.columns.drop(
        labels=[element_column] + [element_type_column] + level_columns + level_weight_columns, errors="ignore"
    )

    # new attributes are created as strings if no type is provided
    try:
        existing_attributes = CaseAndSpaceInsensitiveDict(
            {
                ea.name: ElementAttribute.Types(ea.attribute_type)
                for ea in self.elements.get_element_attributes(dimension_name, hierarchy_name)
            }
        )

    except TM1pyRestException as ex:
        if ex.status_code == 404:
            existing_attributes = set()
        else:
            raise ex

    new_attributes = []
    for attribute_column in attribute_columns:
        if ":" in attribute_column:
            attribute_name, attribute_type = attribute_column.rsplit(":", maxsplit=1)
            attribute_type = self._attribute_type_from_code(attribute_type)

        else:
            attribute_name = attribute_column
            attribute_type = ElementAttribute.Types.STRING

        if attribute_name not in existing_attributes:
            new_attributes.append(ElementAttribute(attribute_name, attribute_type))

        if attribute_name in existing_attributes and update_attribute_types:
            if attribute_type != existing_attributes[attribute_name]:
                self.elements.delete_element_attribute(dimension_name, dimension_name, attribute_name)
                new_attributes.append(ElementAttribute(attribute_name, attribute_type))

    if new_attributes:
        self.elements.add_element_attributes(
            dimension_name=dimension_name, hierarchy_name=hierarchy_name, element_attributes=new_attributes
        )

    # define attributes df with ID + attribute columns.
    id_attribute_cols = [element_column] + list(attribute_columns.values)
    attributes_df: pd.DataFrame = df.loc[:, id_attribute_cols]

    # melt for write structure (ID, Attribute) : Attribute_value
    attributes_df = attributes_df.melt(
        id_vars=element_column,
        value_vars=attribute_columns,
        var_name="}ElementAttributes_" + dimension_name,
        value_name="attribute_value",
    )
    attributes_df.fillna("", inplace=True)

    # drop ':' suffix in attribute column
    attribute_column = "}ElementAttributes_" + dimension_name
    attributes_df[attribute_column] = attributes_df[attribute_column].apply(lambda x: x.rsplit(":", 1)[0])

    # write attributes to cube
    if not attributes_df.empty:
        cell_service = self.get_cell_service()
        # explicitly reference hierarchy if dimension_name != hierarchy_name
        if not case_and_space_insensitive_equals(dimension_name, hierarchy_name):
            attributes_df.iloc[:, 0] = hierarchy_name + ":" + attributes_df.iloc[:, 0].astype(str)
        cell_service.write_dataframe(
            cube_name="}ElementAttributes_" + dimension_name,
            data=attributes_df,
            sum_numeric_duplicates=False,
            use_blob=True,
        )

    if unwind_all:
        self.remove_all_edges(dimension_name=dimension_name, hierarchy_name=hierarchy_name)
    else:
        if unwind_consolidations:
            edges_to_delete = CaseAndSpaceInsensitiveTuplesDict()
            for elem in unwind_consolidations:
                if not self.elements.exists(
                    dimension_name=dimension_name, hierarchy_name=hierarchy_name, element_name=elem
                ):
                    continue

                edges_under_consolidation = self.elements.get_edges_under_consolidation(
                    dimension_name=dimension_name, hierarchy_name=hierarchy_name, consolidation=elem
                )
                edges_to_delete.join(edges_under_consolidation)

            self.elements.delete_edges(
                dimension_name=dimension_name,
                hierarchy_name=hierarchy_name,
                edges=edges_to_delete,
                use_blob=self.is_admin,
            )

    edges = CaseAndSpaceInsensitiveTuplesDict()
    for element_name, *record in df[[element_column, *level_columns, *level_weight_columns]].itertuples(
        index=False
    ):
        levels = record[: len(level_columns)]
        level_weights = record[len(level_columns) :]

        previous_level = element_name
        for level, weight in zip(levels, level_weights):
            if not level:
                continue
            if not isinstance(level, str) and math.isnan(level):
                continue
            if level == previous_level:
                continue

            edges[level, previous_level] = weight
            previous_level = level

    if edges:
        try:
            current_edges = CaseAndSpaceInsensitiveTuplesDict(
                self.elements.get_edges(dimension_name=dimension_name, hierarchy_name=hierarchy_name)
            )
        except TM1pyRestException as ex:
            if ex.status_code == 404:
                current_edges = CaseAndSpaceInsensitiveTuplesDict()
            else:
                raise ex

        edges_to_delete = {(k, v): w for (k, v), w in edges.items() if w != current_edges.get((k, v), w)}
        if edges_to_delete:
            self.elements.delete_edges(
                dimension_name=dimension_name,
                hierarchy_name=hierarchy_name,
                edges=edges_to_delete.keys(),
                use_blob=self.is_admin,
            )

        new_edges = {
            (k, v): w for (k, v), w in edges.items() if (k, v) not in current_edges or w != current_edges[(k, v)]
        }
        if new_edges:
            self.elements.add_edges(dimension_name=dimension_name, hierarchy_name=hierarchy_name, edges=new_edges)

    if hierarchy_sort_order:
        self._implement_hierarchy_sort_order(dimension_name, hierarchy_name, hierarchy_sort_order)

    if delete_orphaned_consolidations:
        all_edges = self.elements.get_edges(dimension_name=dimension_name, hierarchy_name=hierarchy_name)

        parents = CaseAndSpaceInsensitiveSet(parent for parent, _ in all_edges)
        children = CaseAndSpaceInsensitiveSet(child for _, child in all_edges)
        parents_and_children = parents.union(children)

        consolidated_element_names = CaseAndSpaceInsensitiveSet(
            self.elements.get_consolidated_element_names(
                dimension_name=dimension_name, hierarchy_name=hierarchy_name
            )
        )
        orphaned_consolidations = list(consolidated_element_names - parents_and_children)

        if orphaned_consolidations:
            self.elements.delete_elements(
                dimension_name=dimension_name,
                hierarchy_name=hierarchy_name,
                element_names=orphaned_consolidations,
                use_ti=self.is_admin,
            )