Почему я получил ошибку «база данных заблокирована» вместо вставки токена в базу данных ⇐ Python

[Anonymous]

В следующем коде я пытаюсь сохранить токен в базе данных вместе с именем пользователя, но всегда выдает эту ошибку:

Код: Выделить всё

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.

пользователь аутентифицирован, но токен не сохранен в таблице.
Я не знаю, почему он это делает, я все перепробовал Я думал об этом, но это не работает.
если вы можете мне помочь с этим, я буду благодарен.
код:
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"})
)

Я ожидал, что токен будет сохраняться в базе данных после каждой проверки пользователя

Подробнее здесь: https://stackoverflow.com/questions/784 ... the-databa