def to_partial_sql()

in core/lib/sqlparse/models.py [0:0]

• 50 lines of code
• 21 McCabe index (conditional complexity)

    def to_partial_sql(self):
        # Stringify info a format usable in `create table ...`

        def _proc_list(vals: Union[str, List[str]]) -> str:
            # Helper to convert expr list to an expression value-list
            if isinstance(vals, list) and all(isinstance(v, str) for v in vals):
                return "(" + ", ".join(vals) + ")"
            ret = ""
            for v in vals:
                if isinstance(v, list):
                    ret += _proc_list(v)
                else:
                    ret += v
            return ret

        output = f"PARTITION BY {self.full_type}"

        if self.part_type == self.PTYPE_KEY:
            if self.algorithm_for_key is not None:
                output += f" ALGORITHM={self.algorithm_for_key}"
            fields = ", ".join(self.add_quote(f) for f in self.fields_or_expr)
            output += f" ({fields})"
            if self.num_partitions > 1:
                output += f" PARTITIONS {self.num_partitions}"
            return output
        elif self.part_type == self.PTYPE_HASH:
            output += f" ({_proc_list(self.fields_or_expr)})"
            if self.num_partitions > 1:
                output += f" PARTITIONS {self.num_partitions}"
            return output
        elif self.part_type == self.PTYPE_RANGE or self.part_type == self.PTYPE_LIST:
            partitions: List[str] = []
            for pd in self.part_defs:
                name = f"`{pd.pdef_name}`" if pd.pdef_name.isdigit() else pd.pdef_name
                ty = self.TYPE_MAP[pd.pdef_type]
                expr_or_value_list = (
                    _proc_list(pd.pdef_value_list)
                    if isinstance(pd.pdef_value_list, list)
                    else pd.pdef_value_list
                )
                eng = pd.pdef_engine
                if pd.is_tuple:
                    expr_or_value_list = f"({expr_or_value_list})"
                thispart = (
                    f"PARTITION {name} VALUES {ty} {expr_or_value_list} ENGINE {eng}"
                )
                comment = pd.pdef_comment
                if comment is not None:
                    thispart += f" COMMENT {comment}"
                partitions.append(thispart)
            f_or_e = _proc_list(self.fields_or_expr)
            if self.via_nested_expr:
                # PART_EXPR in sqlparse use nestedExpr to acquire this
                # and strips parens so "undo" that
                f_or_e = f"({f_or_e})"
            output += f" {f_or_e} (\n" + ",\n".join(partitions) + ")"
            return output