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

Function get_table_schemas

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

Source from the content-addressed store, hash-verified

32
33
34def get_table_schemas(table_names: List[str] = None, scope="USA") -> str:
35 enums_list = []
36 tables_list = []
37
38 if scope == "USA":
39 enums_list = table_details.get("enums", [])
40 if table_names:
41 for table in table_details['tables']:
42 if table['name'] in table_names:
43 tables_list.append(table)
44 else:
45 tables_list = table_details["tables"]
46 elif scope == "SF":
47 enums_list = sf_table_details["enums"]
48 if table_names:
49 for table in sf_table_details['tables']:
50 if table['name'] in table_names:
51 tables_list.append(table)
52 else:
53 tables_list = sf_table_details["tables"]
54
55 enums_str_set = set()
56 tables_str_list = []
57 for table in tables_list:
58 if scope == "SF":
59
60 tables_str = table['table_creation_query']
61
62 # get all the vars in backticks using regex from tables_str
63 regex = r"`([\s\S]+?)`"
64 matches = re.findall(regex, tables_str)
65 if matches:
66 # add each to enums_str_set
67 for match in matches:
68 enums_str_set.add(match)
69
70 else:
71 tables_str = f"table name: {table['name']}\n"
72 tables_str += f"table description: {table['description']}\n"
73 columns_str_list = []
74 for column in table['columns']:
75 if column.get('description'):
76 columns_str_list.append(f"{column['name']} [{column['type']}] ({column['description']})")
77 if 'custom type' in column['description']:
78 enums_str_set.add(extract_text_from_markdown(column['description']))
79 else:
80 columns_str_list.append(f"{column['name']} [{column['type']}]")
81 tables_str += f"table columns: {', '.join(columns_str_list)}\n"
82 tables_str_list.append(tables_str)
83 tables_description = "\n\n".join(tables_str_list)
84
85 enums_str_list = []
86 for custom_type_str in enums_str_set:
87 custom_type = next((t for t in enums_list if t["type"] == custom_type_str), None)
88 if custom_type:
89 enums_str = f"custom type: {custom_type['type']}\n"
90 enums_str += f"valid values: {', '.join(custom_type['valid_values'])}\n"
91 enums_str_list.append(enums_str)

Calls 1

Tested by

no test coverage detected