| 94 | |
| 95 | |
| 96 | class PGCompleter(Completer): |
| 97 | # keywords_tree: A dict mapping keywords to well known following keywords. |
| 98 | # e.g. 'CREATE': ['TABLE', 'USER', ...], |
| 99 | keywords_tree = get_literals("keywords", type_=dict) |
| 100 | keywords = tuple(set(chain(keywords_tree.keys(), *keywords_tree.values()))) |
| 101 | functions = get_literals("functions") |
| 102 | datatypes = get_literals("datatypes") |
| 103 | reserved_words = set(get_literals("reserved")) |
| 104 | |
| 105 | def __init__(self, smart_completion=True, pgspecial=None, settings=None): |
| 106 | super().__init__() |
| 107 | self.smart_completion = smart_completion |
| 108 | self.pgspecial = pgspecial |
| 109 | self.prioritizer = PrevalenceCounter() |
| 110 | settings = settings or {} |
| 111 | self.signature_arg_style = settings.get("signature_arg_style", "{arg_name} {arg_type}") |
| 112 | self.call_arg_style = settings.get("call_arg_style", "{arg_name: <{max_arg_len}} := {arg_default}") |
| 113 | self.call_arg_display_style = settings.get("call_arg_display_style", "{arg_name}") |
| 114 | self.call_arg_oneliner_max = settings.get("call_arg_oneliner_max", 2) |
| 115 | self.search_path_filter = settings.get("search_path_filter") |
| 116 | self.generate_aliases = settings.get("generate_aliases") |
| 117 | alias_map_file = settings.get("alias_map_file") |
| 118 | if alias_map_file is not None: |
| 119 | self.alias_map = load_alias_map_file(alias_map_file) |
| 120 | else: |
| 121 | self.alias_map = None |
| 122 | self.casing_file = settings.get("casing_file") |
| 123 | self.insert_col_skip_patterns = [ |
| 124 | re.compile(pattern) for pattern in settings.get("insert_col_skip_patterns", [r"^now\(\)$", r"^nextval\("]) |
| 125 | ] |
| 126 | self.generate_casing_file = settings.get("generate_casing_file") |
| 127 | self.qualify_columns = settings.get("qualify_columns", "if_more_than_one_table") |
| 128 | self.asterisk_column_order = settings.get("asterisk_column_order", "table_order") |
| 129 | |
| 130 | keyword_casing = settings.get("keyword_casing", "upper").lower() |
| 131 | if keyword_casing not in ("upper", "lower", "auto"): |
| 132 | keyword_casing = "upper" |
| 133 | self.keyword_casing = keyword_casing |
| 134 | self.name_pattern = re.compile(r"^[_a-z][_a-z0-9\$]*$") |
| 135 | |
| 136 | self.databases = [] |
| 137 | self.dbmetadata = {"tables": {}, "views": {}, "functions": {}, "datatypes": {}} |
| 138 | self.search_path = [] |
| 139 | self.casing = {} |
| 140 | |
| 141 | self.all_completions = set(self.keywords + self.functions) |
| 142 | |
| 143 | def escape_name(self, name): |
| 144 | if name and ((not self.name_pattern.match(name)) or (name.upper() in self.reserved_words) or (name.upper() in self.functions)): |
| 145 | name = '"%s"' % name |
| 146 | |
| 147 | return name |
| 148 | |
| 149 | def escape_schema(self, name): |
| 150 | return "'{}'".format(self.unescape_name(name)) |
| 151 | |
| 152 | def unescape_name(self, name): |
| 153 | """Unquote a string.""" |
no test coverage detected