Executes a SQL query against the database and returns the results
(self, sql_generation_id: str, max_rows: int = 100)
| 427 | |
| 428 | @override |
| 429 | def execute_sql_query(self, sql_generation_id: str, max_rows: int = 100) -> list: |
| 430 | """Executes a SQL query against the database and returns the results""" |
| 431 | sql_generation_service = SQLGenerationService(self.system, self.storage) |
| 432 | try: |
| 433 | results = sql_generation_service.execute(sql_generation_id, max_rows) |
| 434 | except SQLGenerationNotFoundError as e: |
| 435 | raise HTTPException(status_code=404, detail=str(e)) from e |
| 436 | except SQLInjectionError as e: |
| 437 | raise HTTPException(status_code=400, detail=str(e)) from e |
| 438 | except SQLAlchemyError as e: |
| 439 | raise HTTPException(status_code=400, detail=str(e)) from e |
| 440 | except Exception as e: |
| 441 | raise HTTPException(status_code=400, detail=str(e)) from e |
| 442 | return results[1].get("result", []) |
| 443 | |
| 444 | @override |
| 445 | def export_csv_file(self, sql_generation_id: str) -> io.StringIO: |
nothing calls this directly
no test coverage detected