(self, sql: str | Select | TextClause, params=None)
| 2139 | self.con.commit() |
| 2140 | |
| 2141 | def execute(self, sql: str | Select | TextClause, params=None): |
| 2142 | from adbc_driver_manager import Error |
| 2143 | |
| 2144 | if not isinstance(sql, str): |
| 2145 | raise TypeError("Query must be a string unless using sqlalchemy.") |
| 2146 | args = [] if params is None else [params] |
| 2147 | cur = self.con.cursor() |
| 2148 | try: |
| 2149 | cur.execute(sql, *args) |
| 2150 | return cur |
| 2151 | except Error as exc: |
| 2152 | try: |
| 2153 | self.con.rollback() |
| 2154 | except Error as inner_exc: # pragma: no cover |
| 2155 | ex = DatabaseError( |
| 2156 | f"Execution failed on sql: {sql}\n{exc}\nunable to rollback" |
| 2157 | ) |
| 2158 | raise ex from inner_exc |
| 2159 | |
| 2160 | ex = DatabaseError(f"Execution failed on sql '{sql}': {exc}") |
| 2161 | raise ex from exc |
| 2162 | |
| 2163 | def read_table( |
| 2164 | self, |
no test coverage detected