(self, _session: Session, sql, tables: List)
| 818 | return full_dynamic_text |
| 819 | |
| 820 | def generate_assistant_dynamic_sql(self, _session: Session, sql, tables: List): |
| 821 | ds: AssistantOutDsSchema = self.ds |
| 822 | sub_query = [] |
| 823 | result_dict = {} |
| 824 | for table in ds.tables: |
| 825 | if table.name in tables and table.sql: |
| 826 | # sub_query.append({"table": table.name, "query": table.sql}) |
| 827 | result_dict[table.name] = table.sql |
| 828 | sub_query.append({"table": table.name, "query": f'{dynamic_subsql_prefix}{table.name}'}) |
| 829 | if not sub_query: |
| 830 | return None |
| 831 | temp_sql_text = self.generate_with_sub_sql(session=_session, sql=sql, sub_mappings=sub_query) |
| 832 | result_dict['sqlbot_temp_sql_text'] = temp_sql_text |
| 833 | return result_dict |
| 834 | |
| 835 | def build_table_filter(self, session: Session, sql: str, filters: list): |
| 836 | filter = json.dumps(filters, ensure_ascii=False) |
no test coverage detected