MCPcopy Create free account
hub / github.com/caesarHQ/textSQL / get_table_and_enums

Function get_table_and_enums

api/app/api/utils/table_selection/table_details.py:97–158  ·  view source on GitHub ↗
(table_names: List[str] = None, scope="USA")

Source from the content-addressed store, hash-verified

95 return enums_description + "\n\n" + tables_description
96
97def get_table_and_enums(table_names: List[str] = None, scope="USA") -> tuple[str, str]:
98 enums_list = []
99 tables_list = []
100
101 if scope == "USA":
102 enums_list = table_details.get("enums", [])
103 if table_names:
104 for table in table_details['tables']:
105 if table['name'] in table_names:
106 tables_list.append(table)
107 else:
108 tables_list = table_details["tables"]
109 elif scope == "SF":
110 enums_list = sf_table_details["enums"]
111 if table_names:
112 for table in sf_table_details['tables']:
113 if table['name'] in table_names:
114 tables_list.append(table)
115 else:
116 tables_list = sf_table_details["tables"]
117
118 enums_str_set = set()
119 tables_str_list = []
120 for table in tables_list:
121 if scope == "SF":
122
123 tables_str = table['table_creation_query']
124
125 # get all the vars in backticks using regex from tables_str
126 regex = r"`([\s\S]+?)`"
127 matches = re.findall(regex, tables_str)
128 if matches:
129 # add each to enums_str_set
130 for match in matches:
131 enums_str_set.add(match)
132
133 else:
134 tables_str = f"table name: {table['name']}\n"
135 tables_str += f"table description: {table['description']}\n"
136 columns_str_list = []
137 for column in table['columns']:
138 if column.get('description'):
139 columns_str_list.append(f"{column['name']} [{column['type']}] ({column['description']})")
140 if 'custom type' in column['description']:
141 enums_str_set.add(extract_text_from_markdown(column['description']))
142 else:
143 columns_str_list.append(f"{column['name']} [{column['type']}]")
144 tables_str += f"table columns: {', '.join(columns_str_list)}\n"
145 tables_str_list.append(tables_str)
146 tables_description = "\n\n".join(tables_str_list)
147
148 enums_str_list = []
149 for custom_type_str in enums_str_set:
150 custom_type = next((t for t in enums_list if t["type"] == custom_type_str), None)
151 if custom_type:
152 enums_str = f"custom type: {custom_type['type']}\n"
153 enums_str += f"valid values: {', '.join(custom_type['valid_values'])}\n"
154 enums_str_list.append(enums_str)

Callers 1

text_to_sql_with_retryFunction · 0.85

Calls 1

Tested by

no test coverage detected