Database connection problems [login, PyQt5, PySql]

[gradlon93]

Hello Python community,

I'm working on a PyQt5 project and I'm experiencing an issue with a login dialog class that I'm struggling to resolve. Any help and further insights will be greatly appreciated.

Problem Description
In my PyQt5 application, I have a login dialog that should authenticate users against a SQLite database. The dialog works correctly for the first login attempt. However, if the user enters incorrect credentials on the first attempt and then tries again with the correct credentials, the program fails to log in, even with valid credentials. The error message is "Login Failed: Database is not connected."

Code Snippets

LoginDialog Class (Simplified):

class BSLoginDialog(QDialog):
    # Operation codes as in syslog.operations table
    SUCCESSFUL_LOGIN_USER = 7
    SUCCESSFUL_LOGIN_ADMIN = 11
    UNSUCCESSFUL_LOGIN = 8
    CANCEL_LOGIN = 10
    ILLEGAL_ACTIVITY = 14

    def __init__(self):
        super().__init__()

        self.db_manager = BSDatabaseManager()
        self.initialize_db_connection()

        self.setup_ui()
        self.connect_widgets()

    def initialize_db_connection(self):
        if not self.db_manager.connect():
            QMessageBox.warning(
                self,
                "Connection error",
                "There seems to be a connection error, login is temporarily unavailable. Please try again later."
            )
            sys.exit()

    def setup_ui(self):
        # [ Omitted code, for brevity ]
        self.usernameFld = QLineEdit()
        self.usernameFld.setPlaceholderText('Username')
        self.passwordFld = QLineEdit()
        self.passwordFld.setPlaceholderText('Password')
        self.passwordFld.setEchoMode(QLineEdit.Password)
        # ...
        
    def connect_widgets(self):
        self.loginBtn.clicked.connect(self.on_login_clicked)
        self.cancelBtn.clicked.connect(self.on_cancel_clicked)

    def on_login_clicked(self):
        input_password = self.passwordFld.text()
        username = self.usernameFld.text()

        login_successful = self.validate_credentials(username, input_password)

        if login_successful:
            # Successful login
            session_object = LoginCookie(True, self.usr_id, username, self.usr_powers)
            self.register_login(session_object)

        else:
            # Unsuccessful login
            session_object = LoginCookie(False, 2, 'root', 'USER')
            self.register_login(session_object)
            self.login_failed_warning()

        self.db_manager.close()
        if login_successful:
            self.close()

        print(session_object)
        return session_object

    def validate_credentials(self, username, raw_password):
        self.usr_id = None
        self.usr_powers = None
        try:
            query = self.db_manager.execute_query(
                "SELECT usr_id, powers FROM credentials WHERE usr_name = :username",
                {":username": username}
            )

            if query.next():
                self.usr_id = query.value(0)
                self.usr_powers = query.value(1)
                return self.check_password(username, raw_password)
            else:
                return False

        except Exception as e:
            self.login_failed_warning()
            print(f'Login Failed:', e)
            return False

    def register_login(self, cookie: LoginCookie):
        if cookie.state:
            operation_id = self.SUCCESSFUL_LOGIN_ADMIN if cookie.powers == 'ADMIN' else self.SUCCESSFUL_LOGIN_USER
            self.log_activity(cookie.userid, cookie.username, operation_id, str(platform.uname()))
        else:
            self.log_activity(2, 'root', self.UNSUCCESSFUL_LOGIN, str(platform.uname()))
            self.usr_powers = None
            self.usr_id = None
            cookie = None

    def log_activity(self, usr_id, username, operation_id, platform_=None, entity_id=None, entity_type=None):
        platform_ = platform_ if platform_ else str(platform.uname())

        try:
            # Operation info
            self.db_manager.connect()
            operation_query = self.db_manager.execute_query(
                "SELECT operation_phrase FROM operations WHERE operation_id = :operation_id",
                {":operation_id": operation_id}
            )

            if operation_query.next():
                log_message = operation_query.value(0)
            else:
                log_message = 'Operation not found in database.'

            self.db_manager.execute_query(
                """
                INSERT INTO log(usr_id, username, operation_id, log_message, platform, entity_id, entity_type)
                VALUES(:usr_id, :username, :operation_id, :log_message, :platform_, :entity_id, :entity_type);
                """,
                {
                    ":usr_id": usr_id,
                    ":username": username,
                    ":operation_id": operation_id,
                    ":log_message": log_message,
                    ":platform_": platform_,
                    ":entity_id": entity_id,
                    ":entity_type": entity_type
                })

        except Exception as e:
            print(f"Error in log_activity: {e}")
        finally:
            self.db_manager.close()

    def on_cancel_clicked(self):
        self.log_activity(2, 'root', self.CANCEL_LOGIN)
        self.db_manager.close()
        sys.exit()

    def check_password(self, username, input_password):
        try:
            query = self.db_manager.execute_query(
                "SELECT pwd FROM credentials WHERE usr_name = :username",
                {":username": username}
            )

            if query.next():
                stored_hashed_password = query.value(0).encode('utf-8')
                return bcrypt.checkpw(input_password.encode('utf-8'), stored_hashed_password)
            else:
                return False

        except Exception as e:
            self.login_failed_warning()
            print('Login failure:', e)
            return False

    def login_failed_warning(self):
        QMessageBox.warning(self, "Login Failure", "Unable to log in, please try again.")

    def hash_password(self, plain_text_password):
        salt = bcrypt.gensalt()
        hashed_password = bcrypt.hashpw(plain_text_password.encode('utf-8'), salt)
        return hashed_password

DatabaseManager Class

class BSDatabaseManager:
    def __init__(self, db_path='syslog.sqlite'):
        self.db_path = db_path
        self.db = None

    def connect(self):
        self.db = QSqlDatabase.addDatabase("QSQLITE")
        self.db.setDatabaseName(self.db_path)
        return self.db.open()

    def close(self):
        if self.db and self.db.isOpen():
            self.db.close()

    def execute_query(self, query_string, parameters=None):
        if not self.db or not self.db.isOpen():
            raise Exception("Database is not connected")

        query = QSqlQuery(self.db)
        query.prepare(query_string)

        if parameters:
            for key, value in parameters.items():
                query.bindValue(key, value)

        if not query.exec_():
            raise Exception("Query execution failed")

        return query

    def get_all_usernames(self):
        usernames = []

        try:
            query = self.execute_query("SELECT usr_name FROM credentials")
            while query.next():
                usernames.append(query.value(0))
        except Exception as e:
            print(f"Error retrieving usernames: {e}")

        return usernames

Database tables


Specific Issue
The issue seems to be regarding the management of the database connection. After a failed login attempt, it appears that the database connection is not being correctly managed, leading to the error when trying a subsequent login. I've attempted to reset relevant variables and ensure proper connection handling, but the problem persists.

What I tried so far
Ensuring usr_id and usr_powers are reset after each login attempt.
Checking and managing the database connection state more effectively.
Setting "cookie" to None.

Despite these attempts, the issue remains unresolved. I'm looking for advice on how to properly manage the database connection in this context or any other insights that might help solve this problem.

Thank you in advance for your assistance!