(self, sql_query, save_path, max_len, ex_id)
| 83 | |
| 84 | def exec_sql_sf(self, sql_query, save_path, max_len, ex_id): |
| 85 | with self.conns[ex_id].cursor() as cursor: |
| 86 | try: |
| 87 | cursor.execute(sql_query) |
| 88 | column_info = cursor.description |
| 89 | rows = self.get_rows(cursor, max_len) |
| 90 | columns = [desc[0] for desc in column_info] |
| 91 | except Exception as e: |
| 92 | return e |
| 93 | |
| 94 | if not rows: |
| 95 | return "No data found for the specified query.\n" |
| 96 | else: |
| 97 | csv_content = self.get_csv(columns, rows) |
| 98 | if save_path: |
| 99 | with open(save_path, 'w', newline='') as f: |
| 100 | f.write(csv_content) |
| 101 | return 0 |
| 102 | else: |
| 103 | return hard_cut(csv_content, max_len) |
| 104 | |
| 105 | def exec_sql_bq(self, sql_query, save_path, max_len): |
| 106 | bigquery_credential = service_account.Credentials.from_service_account_file("./bigquery_credential.json") |
| 107 | client = bigquery.Client(credentials=bigquery_credential, project=bigquery_credential.project_id) |
no test coverage detected