当前位置：首页 > news >正文

pyqt 操作mysql数据库

news 2026/6/14 20:04:32

import mysql.connector
from datetime import datetime, timezone
from PyQt5.QtCore import pyqtSignal, QObject
import time
import threadingclass VC_DB_Tool(QObject):signalMsg = pyqtSignal(str)def __init__(self, host="localhost", port=3306, database="", username="", password=""):"""初始化MySQL工具类:param host: 数据库主机地址:param port: 数据库端口:param database: 数据库名:param username: 用户名:param password: 密码"""super().__init__()self.host = hostself.port = portself.database = databaseself.username = usernameself.password = passwordself.db = Noneself.cursor = None#
    def set_DBIP(self,db_ip):self.host = db_ip#
    def set_DBPort(self,port):self.port = port#
    #
    def set_DBDatabase(self,database):self.database = database#
    #
    def set_DBUsername(self,username):self.username = username#
    def set_DBPassword(self,password):self.password = password#
    def connect(self):"""连接数据库:return: 连接是否成功"""try:self.db = mysql.connector.connect(host=self.host,port=self.port,database=self.database,user=self.username,password=self.password)self.cursor = self.db.cursor(dictionary=True)self.signalMsg.emit("数据库连接成功")return Trueexcept Exception as e:self.signalMsg.emit(f"数据库连接失败: {str(e)}")return Falsedef disconnect(self):"""断开数据库连接"""if self.db and self.db.is_connected():self.cursor.close()self.db.close()self.signalMsg.emit("数据库连接已断开")def read_table_data(self, table_name):"""读取数据表所有数据:param table_name: 表名:return: 查询结果列表"""if not self.db or not self.db.is_connected():self.signalMsg.emit("数据库未连接")return Nonetry:query = f"SELECT * FROM {table_name}"self.cursor.execute(query)result = self.cursor.fetchall()return resultexcept Exception as e:self.signalMsg.emit(f"读取表数据失败: {str(e)}")return Nonedef query_data(self, table_name, conditions=None):"""根据条件查询数据表数据:param table_name: 表名:param conditions: 查询条件字典，如 {"name": "张三", "age": 25}:return: 查询结果列表"""if not self.db or not self.db.is_connected():self.signalMsg.emit("数据库未连接")return Nonetry:# 构建SQL语句   sql = f"SELECT * FROM {table_name}"if conditions:where_clause = " AND ".join([f"{key}={conditions[key]}" for key in conditions.keys()])sql += f" WHERE {where_clause};"self.signalMsg.emit(f"{str(sql)}")self.cursor.execute(sql, conditions or {})result = self.cursor.fetchall()return resultexcept Exception as e:self.signalMsg.emit(f"条件查询失败: {str(e)}")return Nonedef get_table_count(self, database_name):"""获取某个数据库的数据表数量:param database_name: 数据库名:return: 数据表数量"""if not self.db or not self.db.is_connected():self.signalMsg.emit("数据库未连接")return Nonetry:query = """SELECT COUNT(*) as table_count FROM information_schema.tables WHERE table_schema = %s"""self.cursor.execute(query, (database_name,))result = self.cursor.fetchone()return result['table_count'] if result else Noneexcept Exception as e:self.signalMsg.emit(f"获取表数量失败: {str(e)}")return Nonedef check_table_exists(self, table_name):"""检查表是否存在:param table_name: 表名:return: 是否存在"""if not self.db or not self.db.is_connected():self.signalMsg.emit("数据库未连接")return Falsetry:self.cursor.execute("SHOW TABLES LIKE %s", (table_name,))return self.cursor.fetchone() is not Noneexcept Exception as e:self.signalMsg.emit(f"检查表存在失败: {str(e)}")return False#
    def extract_date_from_table_name(self,table_name):"""从表名中提取日期字符串"""parts = table_name.split('_')if len(parts) >= 4:return parts[-1]return ""def sort_table_names_by_date(self,table_names):"""根据表名中的日期部分对表名进行排序（从最远到最近）"""sorted_tables = sorted(table_names, key=self.extract_date_from_table_name, reverse=False)return sorted_tables#
    def get_all_table_names(self, database_name=None):"""获取数据库中所有表的名称:param database_name: 数据库名，如果为None则使用当前连接的数据库:return: 表名列表"""if not self.db or not self.db.is_connected():self.signalMsg.emit("数据库未连接")return Nonetry:if database_name is None:database_name = self.databasequery = """SELECT table_name FROM information_schema.tables WHERE table_schema = %sORDER BY table_name"""self.cursor.execute(query, (database_name,))result = self.cursor.fetchall()#
            #print(result)# 提取表名列表table_names = [row['TABLE_NAME'] for row in result]# 按日期排序sorted_tables = self.sort_table_names_by_date(table_names)return sorted_tablesexcept Exception as e:self.signalMsg.emit(f"获取表名列表失败: {str(e)}")return None#
#