Execute a SQL statement and return a string representing the results. If the statement returns rows, a string of the results is returned. If the statement returns no rows, an empty string is returned.
(self, command: str, top_k: int = None)
| 202 | return command |
| 203 | |
| 204 | def run_sql(self, command: str, top_k: int = None) -> tuple[str, dict]: |
| 205 | """Execute a SQL statement and return a string representing the results. |
| 206 | |
| 207 | If the statement returns rows, a string of the results is returned. |
| 208 | If the statement returns no rows, an empty string is returned. |
| 209 | """ |
| 210 | with self._engine.connect() as connection: |
| 211 | command = self.parser_to_filter_commands(command) |
| 212 | cursor = connection.execute(text(command)) |
| 213 | if cursor.returns_rows and top_k: |
| 214 | result = cursor.fetchmany(top_k) |
| 215 | return str(result), {"result": result} |
| 216 | if cursor.returns_rows: |
| 217 | result = cursor.fetchall() |
| 218 | return str(result), {"result": result} |
| 219 | return "", {} |
| 220 | |
| 221 | def get_tables_and_views(self) -> List[str]: |
| 222 | inspector = inspect(self._engine) |
no test coverage detected