fully functional, now ui tweaks
This commit is contained in:
41
database.py
41
database.py
@@ -147,7 +147,7 @@ class Database:
|
||||
params.append(workspace_id)
|
||||
|
||||
if search:
|
||||
query += " AND (s.title LIKE ? OR s.slug LIKE ?)"
|
||||
query += " AND (LOWER(s.title) LIKE LOWER(?) OR LOWER(s.slug) LIKE LOWER(?))"
|
||||
params.extend([f"%{search}%", f"%{search}%"])
|
||||
|
||||
query += " ORDER BY s.time_created DESC"
|
||||
@@ -311,18 +311,17 @@ class Database:
|
||||
|
||||
# Copy related records
|
||||
for table, foreign_key in [("message", "session_id"), ("part", "session_id"), ("todo", "session_id"), ("session_share", "session_id")]:
|
||||
cursor.execute(f"SELECT * FROM {table} WHERE {foreign_key} = ?", (sess_id,))
|
||||
try:
|
||||
cursor.execute(f"SELECT * FROM {table} WHERE {foreign_key} = ?", (sess_id,))
|
||||
except sqlite3.OperationalError:
|
||||
continue # Table doesn't exist in this schema version
|
||||
rows = cursor.fetchall()
|
||||
for r in rows:
|
||||
cols = [k for k in r.keys() if k != "id" and k != foreign_key]
|
||||
# Include foreign_key in cols so the new session_id is written
|
||||
cols = [k for k in r.keys() if k != "id"]
|
||||
values = [new_id if k == foreign_key else r[k] for k in cols]
|
||||
col_list = ", ".join(cols)
|
||||
placeholders = ", ".join(["?"] * len(cols))
|
||||
values = [r[col] for col in cols]
|
||||
|
||||
# Set foreign key to new session ID
|
||||
fk_idx = cols.index(foreign_key) if foreign_key in cols else -1
|
||||
if fk_idx >= 0:
|
||||
values[fk_idx] = new_id
|
||||
|
||||
# Generate new ID for tables with id column
|
||||
if "id" in r.keys():
|
||||
@@ -338,6 +337,30 @@ class Database:
|
||||
self.conn.commit()
|
||||
return True, sql_statements, new_session_ids
|
||||
|
||||
def delete_sessions(self, session_ids: List[str]) -> Tuple[bool, str]:
|
||||
"""
|
||||
Permanently delete sessions and all related records.
|
||||
|
||||
Returns:
|
||||
(success, error_message)
|
||||
"""
|
||||
assert self.conn is not None, "Database not connected"
|
||||
cursor = self.conn.cursor()
|
||||
|
||||
try:
|
||||
for sess_id in session_ids:
|
||||
for table, foreign_key in [("message", "session_id"), ("part", "session_id"), ("todo", "session_id"), ("session_share", "session_id")]:
|
||||
try:
|
||||
cursor.execute(f"DELETE FROM {table} WHERE {foreign_key} = ?", (sess_id,))
|
||||
except sqlite3.OperationalError:
|
||||
continue
|
||||
cursor.execute("DELETE FROM session WHERE id = ?", (sess_id,))
|
||||
self.conn.commit()
|
||||
return True, ""
|
||||
except sqlite3.Error as e:
|
||||
self.conn.rollback()
|
||||
return False, str(e)
|
||||
|
||||
def create_backup(self) -> Path:
|
||||
"""Create a timestamped backup of the database."""
|
||||
if not self.db_path.exists():
|
||||
|
||||
Reference in New Issue
Block a user