PyQt5.QtSql使用with块管理数据库连接仍出现未释放警告问题

问题根源 #

警告QSqlDatabasePrivate::removeDatabase: connection 'dbConnection1' is still in use的核心原因是：调用QSqlDatabase.removeDatabase()时，仍有QSqlQuery对象持有该数据库连接的引用。

你的代码中虽然手动调用了query.clear()和del query，但Python的垃圾回收机制不会立刻销毁这些对象，Qt内部可能还保留着对连接的引用，导致removeDatabase执行时连接仍被占用。

修复方案 #

1. 缩小QSqlQuery作用域，让其自动销毁 #

将QSqlQuery的创建限制在最小代码块内，利用Python的作用域规则，在代码块结束后自动回收查询对象，确保调用removeDatabase前所有查询都已销毁。

修改create_table和insert_data方法：

def create_table(self, create_table_sql):
    if not self.table_exists():
        # 仅在需要创建表时创建query，作用域仅限当前代码块
        query = QSqlQuery(self.db_connection)
        if not query.exec_(create_table_sql):
            print("Error creating table:", query.lastError().text())
        else:
            print(f"Table created successfully by {self.db_connection_name}")
    else:
        print(f"Table already exists and does not need to be created by {self.db_connection_name}.")

def insert_data(self, insert_data_sql, values):
    # query创建在函数内部，函数执行完自动销毁
    query = QSqlQuery(self.db_connection)
    query.prepare(insert_data_sql)
    for i, value in enumerate(values):
        query.bindValue(i, value)
    if not query.exec_():
        print("Error inserting data:", query.lastError().text())
    else:
        print(f"Data inserted successfully by {self.db_connection_name}")

2. 修正close方法逻辑 #

• 避免硬编码连接名，使用self.db_connection.connectionName()保证通用性
• 调整操作顺序：先关闭连接，销毁db_connection引用，最后移除数据库连接，确保Qt内部无残留引用

修改close方法：

def close(self):
    if self.db_connection and self.db_connection.isOpen():
        self.db_connection.close()
        conn_name = self.db_connection.connectionName()
        del self.db_connection
        QSqlDatabase.removeDatabase(conn_name)

3. 优化table_exists方法（可选） #

指定只查找用户表，避免系统表干扰，同时确保内部临时查询自动回收：

def table_exists(self):
    tables = self.db_connection.tables(QSqlDatabase.Tables)
    return "example_table" in tables

完整修复后的代码 #

import sys

from PyQt5.QtSql import QSqlDatabase, QSqlQuery
from PyQt5.QtWidgets import QPushButton, QVBoxLayout, QMainWindow, QApplication, QWidget


class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
    
        self.db_name_main = "example.db"
    
        self.setGeometry(100, 100, 570, 600)
        self.setWindowTitle("Database Manipulation")
    
        create_database_button = QPushButton("Create Database ", self)
        create_database_button.clicked.connect(self.test_database_creation)
    
        central_widget = QWidget()
        layout = QVBoxLayout(central_widget)
        layout.addWidget(create_database_button)
    
        self.setCentralWidget(central_widget)
    
    def test_database_creation(self):
        with DatabaseConnectionMaker(self.db_name_main, "dbConnection1") as db_connection1:
            db_connection1.create_table("""CREATE TABLE IF NOT EXISTS example_table (
                   id INTEGER PRIMARY KEY,
                   name TEXT,
                   age INTEGER)""")
    
            db_connection1.insert_data("INSERT INTO example_table (name, age) "
                                       "VALUES (?, ?), (?, ?), (?, ?), (?, ?)",
                                       ["John", 30, "Alice", 25, "Bob", 35, "Eve", 28])
    

class DatabaseConnectionMaker:
    def __init__(self, db_name, db_connection_name):
        self.db_name = db_name
        self.db_connection = None
        self.db_connection_name = db_connection_name

    def connect(self):
        self.db_connection = QSqlDatabase.addDatabase("QSQLITE", self.db_connection_name)
        self.db_connection.setDatabaseName(self.db_name)
        if not self.db_connection.open():
            print("Failed to connect to database:", self.db_connection.lastError().text())
            return False
        return True

    def create_table(self, create_table_sql):
        if not self.table_exists():
            query = QSqlQuery(self.db_connection)
            if not query.exec_(create_table_sql):
                print("Error creating table:", query.lastError().text())
            else:
                print(f"Table created successfully by {self.db_connection_name}")
        else:
            print(f"Table already exists and does not need to be created by {self.db_connection_name}.")

    def table_exists(self):
        tables = self.db_connection.tables(QSqlDatabase.Tables)
        return "example_table" in tables

    def insert_data(self, insert_data_sql, values):
        query = QSqlQuery(self.db_connection)
        query.prepare(insert_data_sql)
        for i, value in enumerate(values):
            query.bindValue(i, value)
        if not query.exec_():
            print("Error inserting data:", query.lastError().text())
        else:
            print(f"Data inserted successfully by {self.db_connection_name}")

    def close(self):
        if self.db_connection and self.db_connection.isOpen():
            self.db_connection.close()
            conn_name = self.db_connection.connectionName()
            del self.db_connection
            QSqlDatabase.removeDatabase(conn_name)

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()
        print(f"Database connection {self.db_connection_name} closed!!!")


if __name__ == "__main__":
    app = QApplication(sys.argv)
    mainWindow = MainWindow()
    mainWindow.show()
    sys.exit(app.exec_())

验证效果 #

运行修复后的代码，点击按钮后将不再出现连接未释放的警告，输出如下：

Table already exists and does not need to be created by dbConnection1.
Data inserted successfully by dbConnection1
Database connection dbConnection1 closed!!!

Process finished with exit code 0

内容的提问来源于stack exchange，提问作者Sam Quintanar