(self, sql=None, params=None, db=None)
| 33 | self.database = db |
| 34 | |
| 35 | def execute(self, sql=None, params=None, db=None): |
| 36 | if sql is None: |
| 37 | return None |
| 38 | |
| 39 | resultRow = [] |
| 40 | with contextlib.\ |
| 41 | closing(mysql.connector.connect(host=str(self.host), |
| 42 | port=int(self.port), |
| 43 | user=str(self.user), |
| 44 | password=str(self.passwd), |
| 45 | db=str(self.database) if not db else db)) as conn: |
| 46 | conn.autocommit = True |
| 47 | with contextlib.closing(conn.cursor(buffered=True)) as cursor: |
| 48 | cursor.execute(sql, params) |
| 49 | try: |
| 50 | if sql.lower().startswith('select') and cursor.rowcount > 0: |
| 51 | # we have more than just the row count/success |
| 52 | resultRow = cursor.fetchall() |
| 53 | except errors.InterfaceError: |
| 54 | # Raised on empty result - DML |
| 55 | resultRow = [] |
| 56 | return resultRow |
| 57 | |
| 58 | def executeSqlFromFile(self, fileName=None): |
| 59 | if fileName is None: |
no test coverage detected