Simple passthrough to SQLAlchemy connectable
(self, sql: str | Select | TextClause | Delete, params=None)
| 1666 | yield self.con |
| 1667 | |
| 1668 | def execute(self, sql: str | Select | TextClause | Delete, params=None): |
| 1669 | """Simple passthrough to SQLAlchemy connectable""" |
| 1670 | from sqlalchemy.exc import SQLAlchemyError |
| 1671 | |
| 1672 | args = [] if params is None else [params] |
| 1673 | if isinstance(sql, str): |
| 1674 | execute_function = self.con.exec_driver_sql |
| 1675 | else: |
| 1676 | execute_function = self.con.execute |
| 1677 | |
| 1678 | try: |
| 1679 | return execute_function(sql, *args) |
| 1680 | except SQLAlchemyError as exc: |
| 1681 | raise DatabaseError(f"Execution failed on sql '{sql}': {exc}") from exc |
| 1682 | |
| 1683 | def read_table( |
| 1684 | self, |
no test coverage detected