Tm1py: tm1.cells.write error when writing to string attribute values

[cubewise-gng]

Describe the bug
Writing string attributes throws an exception when using parameter use_ti=True in the function tm1.cells.write

TM1py exception
File "C:\ProgramData\Anaconda3\lib\site-packages\TM1py\Services\CellService.py", line 453, in write_through_unbound_process raise TM1pyException(f"{len(successes) - sum(successes)} out of {len(successes)} unbound processes failed") TM1py.Exceptions.Exceptions.TM1pyException: 1 out of 1 unbound processes failed

The TI error log "Cell type is string"

To Reproduce
tm1.cells.write(cube_name=cube_name, cellset_as_dict=cellset_attributes, use_ti=True)

Version
TM1py 1.5.0

[rkvinoth]

@MariusWirtz ,

Here is a solution but not sure if this is the most efficient way to do this. I'm using a DTYPE check here in the TI for every cellset item.

    @require_admin
    @manage_transaction_log
    def write_through_unbound_process(self, cube_name: str, cellset_as_dict: Dict, increment: bool = False,
                                      sandbox_name: str = None, **kwargs):
        if sandbox_name:
            raise NotImplementedError("Function does not support writing to sandboxes yet")

        successes = list()
        statements = list()

        from TM1py import CubeService
        cube_service = CubeService(self._rest)
        measure_dimension = cube_service.get(cube_name).dimensions[-1]
        function_str = f"{'CellIncrementN(' if increment else 'CellPutN('}"
        for coordinates, value in cellset_as_dict.items():
            value_str = f'{value}' if isinstance(value, str) else str(value)

            statement = f"IF(DTYPE('{measure_dimension}', '{coordinates[-1]}') @= 'N');\n"
            try:
                float(value)
                statement = "".join([
                    statement,
                    function_str,
                    value_str,
                    f",'{cube_name}',",
                    ",".join(f"'{element}'" for element in coordinates),
                    ");\n"])
            except ValueError:
                statement = "".join([statement, "'This is a temp line.';\n"])
            statement=  "".join([statement, "ELSE;\n"])
            statement = "".join([
                statement,
                'CellPutS(',
                f"'{value_str}'",
                f",'{cube_name}',",
                ",".join(f"'{element}'" for element in coordinates),
                ");\n",
                "ENDIF;\n"])
            statements.append(statement)

        chunk = list()
        for n, statement in enumerate(statements):
            chunk.append(statement)
            if n > 0 and n % Process.MAX_STATEMENTS == 0:
                process = Process(name="", prolog_procedure="\r".join(chunk))
                success, _, _ = self.execute_unbound_process(process, **kwargs)
                successes.append(success)
                chunk = list()

        process = Process(name="", prolog_procedure="\r".join(chunk))
        success, _, _ = self.execute_unbound_process(process, **kwargs)
        successes.append(success)

        if not all(successes):
            raise TM1pyException(f"{len(successes) - sum(successes)} out of {len(successes)} unbound processes failed")

[scrumthing]

I think that is the only way to check for a string within a process.
Alternatively you could add a parameter to write numeric only to avoid the check and gain some performance.

[rkvinoth]

The previous approach has a bug. This version seems to work faster.

1. Checking the element type and using the respective TM1 function
2. skipping any consolidated elements

    @require_admin
    @manage_transaction_log
    def write_through_unbound_process(self, cube_name: str, cellset_as_dict: Dict, increment: bool = False,
                                      sandbox_name: str = None, **kwargs):
        if sandbox_name:
            raise NotImplementedError("Function does not support writing to sandboxes yet")

        successes = list()
        statements = list()

        from TM1py import CubeService
        from TM1py import ElementService
        cube_service = CubeService(self._rest)
        element_service = ElementService(self._rest)
        measure_dimension = cube_service.get(cube_name).dimensions[-1]
        measure_dimension_elements = element_service.get_element_types(dimension_name=measure_dimension,
                                                                       hierarchy_name=measure_dimension,
                                                                       skip_consolidations=True)

        for coordinates, value in cellset_as_dict.items():
            element_type = measure_dimension_elements.get(coordinates[-1])
            if element_type == "Numeric":
                function_str = f"{'CellIncrementN(' if increment else 'CellPutN('}"
                value_str = f"{value}"
            elif element_type == "String":
                function_str = 'CellPutS('
                value_str = f"'{value}'"
            else:
                continue
            statement = "".join([
                function_str,
                value_str,
                f",'{cube_name}',",
                ",".join(f"'{element}'" for element in coordinates),
                ");\n"])
            statements.append(statement)

        chunk = list()
        for n, statement in enumerate(statements):
            chunk.append(statement)
            if n > 0 and n % Process.MAX_STATEMENTS == 0:
                process = Process(name="", prolog_procedure="\r".join(chunk))
                success, _, _ = self.execute_unbound_process(process, **kwargs)
                successes.append(success)
                chunk = list()

        process = Process(name="", prolog_procedure="\r".join(chunk))
        success, _, _ = self.execute_unbound_process(process, **kwargs)
        successes.append(success)

        if not all(successes):
            raise TM1pyException(f"{len(successes) - sum(successes)} out of {len(successes)} unbound processes failed")

[MariusWirtz]

@rkvinoth thanks for the proposed solution.

I created a PR based on your proposal:
https://github.com/cubewise-code/tm1py/pull/446

Please review and approve my changes :)

[MariusWirtz]

Fixed with #446