def __call__()

in core/maxframe/dataframe/indexing/setitem.py [0:0]

• 58 lines of code
• 19 McCabe index (conditional complexity)

    def __call__(self, target: DataFrame, value):
        raw_target = target

        inputs = [target]
        if np.isscalar(value):
            value_dtype = np.array(value).dtype
        elif self._is_scalar_tensor(value):
            inputs.append(value)
            value_dtype = value.dtype
        else:
            if isinstance(value, (pd.Series, SERIES_TYPE)):
                value = asseries(value)
                value_dtype = value.dtype
            elif isinstance(value, (pd.DataFrame, DATAFRAME_TYPE)):
                if len(self.indexes) != value.shape[1]:  # pragma: no cover
                    raise ValueError("Columns must be same length as key")

                value = asframe(value)
                value_dtype = pd.Series(list(value.dtypes), index=self.indexes)
            elif is_list_like(value) or isinstance(value, TENSOR_TYPE):
                # convert to numpy to get actual dim and shape
                if is_list_like(value):
                    value = np.array(value)

                if value.ndim == 1:
                    value = asseries(value, index=target.index)
                    value_dtype = value.dtype
                else:
                    if len(self.indexes) != value.shape[1]:  # pragma: no cover
                        raise ValueError("Columns must be same length as key")

                    value = asframe(value, index=target.index)
                    value_dtype = pd.Series(list(value.dtypes), index=self.indexes)
            else:  # pragma: no cover
                raise TypeError(
                    "Wrong value type, could be one of scalar, Series or tensor"
                )

            if target.shape[0] == 0:
                # target empty, reindex target first
                target = target.reindex(value.index)
                inputs[0] = target
            elif value.index_value.key != target.index_value.key:
                # need reindex when target df is not empty and index different
                value = value.reindex(target.index)
            inputs.append(value)

        index_value = target.index_value
        dtypes = target.dtypes.copy(deep=True)

        try:
            dtypes.loc[self.indexes] = value_dtype
        except KeyError:
            # when some index not exist, try update one by one
            if isinstance(value_dtype, pd.Series):
                for idx in self.indexes:
                    dtypes.loc[idx] = value_dtype.loc[idx]
            else:
                for idx in self.indexes:
                    dtypes.loc[idx] = value_dtype

        columns_value = parse_index(dtypes.index, store_data=True)
        ret = self.new_dataframe(
            inputs,
            shape=(target.shape[0], len(dtypes)),
            dtypes=dtypes,
            index_value=index_value,
            columns_value=columns_value,
        )
        raw_target.data = ret.data