(self, sql: str | Select | TextClause, params=None)
| 2735 | cur.close() |
| 2736 | |
| 2737 | def execute(self, sql: str | Select | TextClause, params=None): |
| 2738 | from sqlite3 import Error |
| 2739 | |
| 2740 | if not isinstance(sql, str): |
| 2741 | raise TypeError("Query must be a string unless using sqlalchemy.") |
| 2742 | args = [] if params is None else [params] |
| 2743 | cur = self.con.cursor() |
| 2744 | try: |
| 2745 | cur.execute(sql, *args) |
| 2746 | return cur |
| 2747 | except Error as exc: |
| 2748 | try: |
| 2749 | self.con.rollback() |
| 2750 | except Error as inner_exc: # pragma: no cover |
| 2751 | ex = DatabaseError( |
| 2752 | f"Execution failed on sql: {sql}\n{exc}\nunable to rollback" |
| 2753 | ) |
| 2754 | raise ex from inner_exc |
| 2755 | |
| 2756 | ex = DatabaseError(f"Execution failed on sql '{sql}': {exc}") |
| 2757 | raise ex from exc |
| 2758 | |
| 2759 | @staticmethod |
| 2760 | def _query_iterator( |
no test coverage detected