def get_column_matches()

in ossdbtoolsservice/language/completion/pgcompleter.py [0:0]

• 53 lines of code
• 32 McCabe index (conditional complexity)

    def get_column_matches(self, suggestion, word_before_cursor):
        tables = suggestion.table_refs
        do_qualify = suggestion.qualifiable and {'always': True, 'never': False, 'if_more_than_one_table': len(tables) > 1}[self.qualify_columns]

        def qualify(col, tbl): return ((tbl + '.' + self.case(col)) if do_qualify else self.case(col))     # noqa
        self._log(False, "Completion column scope: %r", tables)
        scoped_cols = self.populate_scoped_cols(tables, suggestion.local_tables)

        def make_cand(name, ref):
            synonyms = (name, generate_alias(self.case(name)))
            return Candidate(qualify(name, ref), 0, 'column', synonyms, schema=None)

        def flat_cols():
            return [make_cand(c.name, t.ref) for t, cols in scoped_cols.items() for c in cols]
        if suggestion.require_last_table:
            # require_last_table is used for 'tb11 JOIN tbl2 USING (...' which should
            # suggest only columns that appear in the last table and one more
            ltbl = tables[-1].ref
            other_tbl_cols = set(
                c.name for t, cs in scoped_cols.items() if t.ref != ltbl for c in cs)
            scoped_cols = {
                t: [col for col in cols if col.name in other_tbl_cols]
                for t, cols in scoped_cols.items()
                if t.ref == ltbl
            }
        lastword = last_word(word_before_cursor, include='most_punctuations')
        if lastword == '*':
            if suggestion.context == 'insert':
                def filter(col):
                    if not col.has_default:
                        return True
                    return not any(
                        p.match(col.default)
                        for p in self.insert_col_skip_patterns
                    )
                scoped_cols = {
                    t: [col for col in cols if filter(col)] for t, cols in scoped_cols.items()
                }
            if self.asterisk_column_order == 'alphabetic':
                for cols in scoped_cols.values():
                    cols.sort(key=operator.attrgetter('name'))
            if (lastword != word_before_cursor and len(tables) == 1 and word_before_cursor[-len(lastword) - 1] == '.'):
                # User typed x.*; replicate "x." for all columns except the
                # first, which gets the original (as we only replace the "*"")
                sep = ', ' + word_before_cursor[:-1]
                collist = sep.join(self.case(c.completion)
                                   for c in flat_cols())
            else:
                collist = ', '.join(qualify(c.name, t.ref)
                                    for t, cs in scoped_cols.items() for c in cs)

            return [Match(
                completion=PGCompletion(
                    collist,
                    -1,
                    display_meta='columns',
                    display='*'
                ),
                priority=(1, 1, 1)
            )]

        return self.find_matches(word_before_cursor, flat_cols(), meta='column')