Method get_column_matches

pgcli/pgcompleter.py:487–545

(self, suggestion, word_before_cursor)

        return [m.completion for m in matches]

    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)

        _logger.debug("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), alias_map=self.alias_map))
            return Candidate(qualify(name, ref), 0, "column", synonyms)

        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 = {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=Completion(collist, -1, display_meta="columns", display="*"),
                    priority=(1, 1, 1),
                )
            ]