(self, sql: Optional[str], dbID: Optional[str])
| 34 | self.dbMap = {} |
| 35 | |
| 36 | def operate(self, sql: Optional[str], dbID: Optional[str]) -> [int, Any]: |
| 37 | sqlType = get_db_operation_class(sql=sql) |
| 38 | |
| 39 | db = self.dbMap[dbID] |
| 40 | cursor = db.cursor() |
| 41 | |
| 42 | if sqlType == SQL_class.UPDATE: |
| 43 | cursor.execute(sql) |
| 44 | db.commit() |
| 45 | updated_rows = cursor.rowcount |
| 46 | return SQL_class.UPDATE, updated_rows |
| 47 | elif sqlType == SQL_class.DELETE: |
| 48 | cursor.execute(sql) |
| 49 | db.commit() |
| 50 | deleted_rows = cursor.rowcount |
| 51 | return SQL_class.DELETE, deleted_rows |
| 52 | elif sqlType == SQL_class.INSERT: |
| 53 | cursor.execute(sql) |
| 54 | db.commit() |
| 55 | new_row_id = cursor.lastrowid |
| 56 | return SQL_class.INSERT, new_row_id |
| 57 | elif sqlType == SQL_class.SELECT: |
| 58 | cursor.execute(sql) |
| 59 | results = cursor.fetchall() |
| 60 | cursor.close() |
| 61 | db.close() |
| 62 | dataframe = pd.DataFrame(results, columns=[i[0] for i in cursor.description]) # 根据你的列名进行修改 |
| 63 | return SQL_class.SELECT, dataframe |
| 64 | else: |
| 65 | return -1, -1 |
| 66 | |
| 67 | def add_db(self, dbAddress: Optional[str], dbPort: Optional[str], userName: Optional[str], password: Optional[str], dbName: Optional[str], dbID: Optional[str]) -> str: |
| 68 | db = pymysql.connect(host=dbAddress, port=int(dbPort), user=userName, password=password, database=dbName) |
no test coverage detected