57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
import sqlite3
|
|
from dataclasses import dataclass
|
|
from typing import Optional, List
|
|
|
|
# --- DATACLASSES ---
|
|
|
|
@dataclass
|
|
class MessageModel:
|
|
id: Optional[int] = None
|
|
conversation_id: Optional[int] = None
|
|
sender_id: Optional[int] = None
|
|
content: Optional[str] = None
|
|
created_at: Optional[str] = None
|
|
|
|
class Messages:
|
|
def __init__(self, db_path="instance/app_database.db"):
|
|
self.db_path = db_path
|
|
self._create_tables()
|
|
|
|
def _create_tables(self):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
conversation_id INTEGER,
|
|
sender_id INTEGER,
|
|
content TEXT NOT NULL,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (conversation_id) REFERENCES conversations(id) ON DELETE CASCADE,
|
|
FOREIGN KEY (sender_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
""")
|
|
conn.commit()
|
|
|
|
def send(self, conversation_id: int, sender_id: int, content: str) -> int:
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"INSERT INTO messages (conversation_id, sender_id, content) VALUES (?, ?, ?)",
|
|
(conversation_id, sender_id, content)
|
|
)
|
|
return cursor.lastrowid
|
|
|
|
def get_history(self, conversation_id: int, limit: int = 50) -> List[MessageModel]:
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT * FROM messages WHERE conversation_id = ? ORDER BY created_at DESC LIMIT ?",
|
|
(conversation_id, limit)
|
|
)
|
|
return [MessageModel(**dict(row)) for row in cursor.fetchall()]
|
|
|
|
def delete_message(self, message_id: int):
|
|
with sqlite3.connect(self.db_path) as conn:
|
|
conn.execute("DELETE FROM messages WHERE id = ?", (message_id,)) |