117 lines
3.2 KiB
Python
Executable File
117 lines
3.2 KiB
Python
Executable File
"""
|
|
Thread table class for database operations.
|
|
|
|
This module provides the ThreadTable class that handles thread-specific
|
|
database operations for storing and retrieving Thread objects.
|
|
"""
|
|
|
|
import sqlite3
|
|
from typing import Optional
|
|
from ..Datamodel.Thread import Thread
|
|
|
|
|
|
class ThreadTable:
|
|
"""
|
|
Table class for managing thread data in the database.
|
|
|
|
Attributes:
|
|
connection: SQLite database connection
|
|
cursor: Database cursor for executing queries
|
|
"""
|
|
|
|
def __init__(self, connection: sqlite3.Connection):
|
|
"""
|
|
Initialize the thread table with a database connection.
|
|
|
|
Args:
|
|
connection: SQLite database connection
|
|
"""
|
|
self.connection = connection
|
|
self.cursor = self.connection.cursor()
|
|
|
|
def create(self):
|
|
"""Create the threads table if it doesn't exist."""
|
|
self.cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS threads (
|
|
id TEXT PRIMARY KEY,
|
|
title TEXT NOT NULL,
|
|
board_id TEXT,
|
|
view_count INTEGER,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
''')
|
|
self.connection.commit()
|
|
|
|
def exists(self, thread_id: str) -> bool:
|
|
"""
|
|
Check if a thread exists in the database.
|
|
|
|
Args:
|
|
thread_id: Thread ID to check
|
|
|
|
Returns:
|
|
True if thread exists, False otherwise
|
|
"""
|
|
self.cursor.execute('SELECT 1 FROM threads WHERE id = ?', (thread_id,))
|
|
return self.cursor.fetchone() is not None
|
|
|
|
def save(self, thread: Thread, board_name: str = None) -> bool:
|
|
"""
|
|
Save a thread to the database.
|
|
|
|
Args:
|
|
thread: Thread object to save
|
|
board_name: Board name that this thread belongs to
|
|
|
|
Returns:
|
|
True if thread was saved, False if it already exists
|
|
"""
|
|
if self.exists(thread.id):
|
|
return False
|
|
|
|
self.cursor.execute('''
|
|
INSERT INTO threads (id, title, board_id, view_count)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (
|
|
thread.id,
|
|
thread.title,
|
|
board_name,
|
|
thread.view_count
|
|
))
|
|
|
|
self.connection.commit()
|
|
return True
|
|
|
|
def load(self, thread_id: str) -> Optional[Thread]:
|
|
"""
|
|
Load a thread from the database by ID.
|
|
|
|
Args:
|
|
thread_id: Thread ID to load
|
|
|
|
Returns:
|
|
Thread object if found, None otherwise
|
|
"""
|
|
self.cursor.execute('''
|
|
SELECT id, title, board_id, view_count
|
|
FROM threads WHERE id = ?
|
|
''', (thread_id,))
|
|
|
|
row = self.cursor.fetchone()
|
|
if row is None:
|
|
return None
|
|
|
|
return Thread(
|
|
id=row[0],
|
|
title=row[1],
|
|
board_id=row[2],
|
|
view_count=row[3]
|
|
)
|
|
|
|
def commit(self):
|
|
"""Commit pending changes to the database."""
|
|
self.connection.commit()
|
|
|
|
def close(self):
|
|
"""Close the cursor."""
|
|
self.cursor.close() |