本地认证成功但存储Token时触发SQLite database is locked错误

开发认证系统时，尝试将用户Token与用户名存入SQLite数据库，出现以下错误：

authenticating... 
2024-05-10 01:41:58,350 - INFO - authenticating locally... 
2024-05-10 01:42:03,663 - ERROR - An unexpected error occurred: database is locked 
2024-05-10 01:42:03,663 - INFO - Database connection closed. 
2024-05-10 01:42:03,663 - INFO - User admin authenticated successfully.

用户本地认证成功，但Token未写入Sessions表，请求解决。

相关代码 #

Authenticator.py #

import sqlite3, secrets, logging

# to provide password hashing utilites
from passlib.hash import argon2
from datahandler import checkInternet

from AddToken import add_token_to_database

# Configures the logging system to display all logs with a severity level of INFO or higher.
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
# Creates a logger with the name ‘authenticator’for logging messages within this module.
logger = logging.getLogger(__name__)


class DatabaseManager:
    def __init__(self, db_path):
        # Establishes a connection to the SQLite database at the specified path.
        self.conn = sqlite3.connect(db_path)
        # Creates a cursor object to execute SQL commands.
        self.cursor = self.conn.cursor()

    def close(self):
        # Closes the SQLite database connection.
        self.conn.close()


class Authenticator:
    def __init__(self, db_manager):
        self.db_manager = db_manager

    async def authenticate_online(self, username, password):
        # Placeholder for online authentication logic
        return False, None

    def authenticate_locally(self, username, password):
        logger.info("authenticating locally...")
        try:
            self.db_manager.cursor.execute(
                "SELECT Password FROM Users WHERE Username = ?", (username,)
            )
            result = self.db_manager.cursor.fetchone()

            if result is None:
                logger.error(f"No user found with the username: {username}")
                return False, None

            hashed_password = result[0]
            if hashed_password and argon2.verify(password, hashed_password):
                token = create_session_token()
                add_token_to_database(username, token)
                logger.info(f"User {username} authenticated successfully.")
                return True, token
            else:
                logger.error(f"Password verification failed for user: {username}")
                return False, None
        except sqlite3.Error as e:
            logger.error(f"SQLite error during local authentication: {e}")
            return False, None
        except ValueError as e:
            logger.error(f"Malformed Argon2 hash, or other library error: {e}")
            return False, None

    async def authenticate(self, username, password):
        print("authenticating...")
        if checkInternet("www.google.com", 3):
            # online authentication
            success, token = await self.authenticate_online(username, password)
        else:
            # offline authentication
            success, token = self.authenticate_locally(username, password)
            if success:
                # Ensures that a valid token is returned, either a new one or an existing valid session token.
                token = token or self.get_valid_session_token(username)
        return success, token

    # retrieves a valid session token for the user if it exists.
    def get_valid_session_token(self, username):
        self.db_manager.cursor.execute(
            "SELECT Token FROM Sessions WHERE Username = ?", (username,)
        )
        result = self.db_manager.cursor.fetchone()
        if result:
            return result[0]
        return None


# function to generate a secure session token using the secrets module
def create_session_token():
    return secrets.token_urlsafe(16)

AddToken.py #

import sqlite3, logging


logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def add_token_to_database(username, token, db_path="DB.db"):
    try:
        with sqlite3.connect(db_path) as conn:
            cursor = conn.cursor()
            cursor.execute(
                """
                INSERT OR REPLACE INTO Sessions (Username, Token) VALUES (?, ?)
                """,
                (username, token),
            )
            conn.commit()
            logger.info("Session token inserted/updated successfully for user: %s", username)
    except sqlite3.IntegrityError as e:
        logger.error(f"Integrity error when adding token: {e}")
    except Exception as e:
        logger.error(f"An unexpected error occurred: {e}")
    finally:
        logger.info("Database connection closed.")

loginHandler.py #

import json, logging
from Authenticator import DatabaseManager, Authenticator


db_path = "./DB.db"
db_manager = DatabaseManager(db_path)
authenticator = Authenticator(db_manager)

# Initialize logger
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


# An asynchronous function that listens for login requests over a WebSocket connection
# and It decodes the JSON message, extracts the username and password, and attempts to authenticate the user
# If authentication is successful, it sends back a JSON response with a success flag and a session token.
async def handle_login(websocket, path):
    async for message in websocket:
        try:
            data = json.loads(message)
            if data.get("action") == "login":
                username = data["username"]
                password = data["password"]
                authenticated, token = await authenticator.authenticate(
                    username, password
                )
                if authenticated:
                    await websocket.send(json.dumps({"success": True, "token": token}))
                else:
                    await websocket.send(json.dumps({"success": False}))
        except json.JSONDecodeError:
            await websocket.send(
                json.dumps({"success": False, "error": "Invalid JSON"})
            )
        except Exception as e:
            logger.error("Login handling error: %s", e)
            await websocket.send(
                json.dumps({"success": False, "error": "Internal server error"})
            )

问题根源 #

SQLite是文件级数据库，同一时间仅允许一个写操作。当前代码存在两个并发连接：

1. loginHandler.py初始化的DatabaseManager保持了持久连接（未关闭）
2. add_token_to_database每次调用都会新建连接尝试写入

当authenticate_locally调用add_token_to_database时，第一个连接仍处于打开状态，导致写操作被数据库锁阻塞。

修复步骤 #

1. 修改add_token_to_database复用现有连接 #

修改AddToken.py，让函数接收已有的数据库连接对象，避免新建连接：

import logging

logger = logging.getLogger(__name__)
def add_token_to_database(username, token, conn):
    try:
        cursor = conn.cursor()
        cursor.execute(
            """
            INSERT OR REPLACE INTO Sessions (Username, Token) VALUES (?, ?)
            """,
            (username, token),
        )
        conn.commit()
        logger.info("Session token inserted/updated successfully for user: %s", username)
    except sqlite3.IntegrityError as e:
        logger.error(f"Integrity error when adding token: {e}")
    except Exception as e:
        logger.error(f"An unexpected error occurred: {e}")

2. 更新Authenticator中的调用逻辑 #

在Authenticator.py的authenticate_locally方法中，传入已有的数据库连接：

# 替换原add_token_to_database(username, token)
add_token_to_database(username, token, self.db_manager.conn)

3. 可选：优化数据库连接管理 #

给DatabaseManager添加事务管理方法，确保操作的一致性：

class DatabaseManager:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()

    def close(self):
        self.conn.close()

    def commit(self):
        self.conn.commit()

    def rollback(self):
        self.conn.rollback()

这样所有数据库操作都复用同一个连接，避免了SQLite的锁冲突问题，Token就能正常写入Sessions表。

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