def join_partitions()

in src/ab/plugins/db/odps_helper.py [0:0]

• 22 lines of code
• 14 McCabe index (conditional complexity)

    def join_partitions(partitions: list) -> str:
        """
        join partitions into where conditions
        input:
            None
            []
            ['p1=v1/p2=v2', 'p1=v3/p2=v4'] or ['p1=v1,p2=v2', 'p1=v3,p2=v4']
        returns:
            None: not a partitioned table
            '': no data
            '((p1='v1' and p2='v2') or (p1='v3' and p2='v4'))'
        """
        if partitions is None:
            return None
        elif not partitions:
            return ''
        else:
            # partitions: ['p1=v1/p2=v2', yyy] ->
            #     p: 'p1=v1/p2=v2' ->
            #     kv_pairs: ['p1=v1', 'p2=v2'] ->
            #         kv_pair: 'p1=v1' ->
            #             k: p1   v: 'v1' ->
            #     quoted_kv_pairs: ["p1='v1'", "p2='v2'"] ->
            # partition_strings: ["p1='v1' and p2='v2'"] ->
            # return ((p1='v1' and p2='v2') or (yyy))

            partition_strings = []
            for p in partitions:
                # odps does not allow '/' or ',' in partition keys, just split
                if '/' in p:
                    kv_pairs = p.split('/')
                elif ',' in p:
                    kv_pairs = p.split(',')
                else:
                    kv_pairs = [p, ]

                quoted_kv_pairs = []
                for kv_pair in kv_pairs:
                    k, v = kv_pair.split('=')
                    if "'" not in v and '"' not in v:
                        # convert string partition key. bigint key compatible
                        v = "'{v}'".format(v=v)
                    quoted_kv_pairs.append('{k}={v}'.format(k=k, v=v))
                partition_strings.append(' and '.join(quoted_kv_pairs))

            return '((' + ') or ('.join(partition_strings) + '))'