(self, sql, param=())
| 294 | return "error: " + str(ex) |
| 295 | |
| 296 | def execute(self, sql, param=()): |
| 297 | # 执行SQL语句返回受影响行 |
| 298 | self.write_lock() |
| 299 | self.__GetConn() |
| 300 | try: |
| 301 | result = self.__DB_CONN.execute(sql, self.__to_tuple(param)) |
| 302 | self.__DB_CONN.commit() |
| 303 | self.rm_lock() |
| 304 | return result.rowcount |
| 305 | except Exception as ex: |
| 306 | return "error: " + str(ex) |
| 307 | |
| 308 | # 是否有锁 |
| 309 | def is_lock(self): |
no test coverage detected