my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import logging
#2	import sqlite3
#3	import threading
#4	import uuid
#5	from typing import Any, Dict, List, Optional
#6
#7	logger = logging.getLogger(__name__)
#8
#9
#10	class SQLiteManager:
#11	def __init__(self, db_path: str = ":memory:"):
#12	self.db_path = db_path
#13	self.connection = sqlite3.connect(self.db_path, check_same_thread=False)
#14	self._lock = threading.Lock()
#15	self._migrate_history_table()
#16	self._create_history_table()
#17
#18	def _migrate_history_table(self) -> None:
#19	"""
#20	If a pre-existing history table had the old group-chat columns,
#21	rename it, create the new schema, copy the intersecting data, then
#22	drop the old table.
#23	"""
#24	with self._lock:
#25	try:
#26	# Start a transaction
#27	self.connection.execute("BEGIN")
#28	cur = self.connection.cursor()
#29
#30	cur.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='history'")
#31	if cur.fetchone() is None:
#32	self.connection.execute("COMMIT")
#33	return # nothing to migrate
#34
#35	cur.execute("PRAGMA table_info(history)")
#36	old_cols = {row[1] for row in cur.fetchall()}
#37
#38	expected_cols = {
#39	"id",
#40	"memory_id",
#41	"old_memory",
#42	"new_memory",
#43	"event",
#44	"created_at",
#45	"updated_at",
#46	"is_deleted",
#47	"actor_id",
#48	"role",
#49	}
#50
#51	if old_cols == expected_cols:
#52	self.connection.execute("COMMIT")
#53	return
#54
#55	logger.info("Migrating history table to new schema (no convo columns).")
#56
#57	# Clean up any existing history_old table from previous failed migration
#58	cur.execute("DROP TABLE IF EXISTS history_old")
#59
#60	# Rename the current history table
#61	cur.execute("ALTER TABLE history RENAME TO history_old")
#62
#63	# Create the new history table with updated schema
#64	cur.execute(
#65	"""
#66	CREATE TABLE history (
#67	id TEXT PRIMARY KEY,
#68	memory_id TEXT,
#69	old_memory TEXT,
#70	new_memory TEXT,
#71	event TEXT,
#72	created_at DATETIME,
#73	updated_at DATETIME,
#74	is_deleted INTEGER,
#75	actor_id TEXT,
#76	role TEXT
#77	)
#78	"""
#79	)
#80
#81	# Copy data from old table to new table
#82	intersecting = list(expected_cols & old_cols)
#83	if intersecting:
#84	cols_csv = ", ".join(intersecting)
#85	cur.execute(f"INSERT INTO history ({cols_csv}) SELECT {cols_csv} FROM history_old")
#86
#87	# Drop the old table
#88	cur.execute("DROP TABLE history_old")
#89
#90	# Commit the transaction
#91	self.connection.execute("COMMIT")
#92	logger.info("History table migration completed successfully.")
#93
#94	except Exception as e:
#95	# Rollback the transaction on any error
#96	self.connection.execute("ROLLBACK")
#97	logger.error(f"History table migration failed: {e}")
#98	raise
#99
#100	def _create_history_table(self) -> None:
#101	with self._lock:
#102	try:
#103	self.connection.execute("BEGIN")
#104	self.connection.execute(
#105	"""
#106	CREATE TABLE IF NOT EXISTS history (
#107	id TEXT PRIMARY KEY,
#108	memory_id TEXT,
#109	old_memory TEXT,
#110	new_memory TEXT,
#111	event TEXT,
#112	created_at DATETIME,
#113	updated_at DATETIME,
#114	is_deleted INTEGER,
#115	actor_id TEXT,
#116	role TEXT
#117	)
#118	"""
#119	)
#120	self.connection.execute("COMMIT")
#121	except Exception as e:
#122	self.connection.execute("ROLLBACK")
#123	logger.error(f"Failed to create history table: {e}")
#124	raise
#125
#126	def add_history(
#127	self,
#128	memory_id: str,
#129	old_memory: Optional[str],
#130	new_memory: Optional[str],
#131	event: str,
#132	*,
#133	created_at: Optional[str] = None,
#134	updated_at: Optional[str] = None,
#135	is_deleted: int = 0,
#136	actor_id: Optional[str] = None,
#137	role: Optional[str] = None,
#138	) -> None:
#139	with self._lock:
#140	try:
#141	self.connection.execute("BEGIN")
#142	self.connection.execute(
#143	"""
#144	INSERT INTO history (
#145	id, memory_id, old_memory, new_memory, event,
#146	created_at, updated_at, is_deleted, actor_id, role
#147	)
#148	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
#149	""",
#150	(
#151	str(uuid.uuid4()),
#152	memory_id,
#153	old_memory,
#154	new_memory,
#155	event,
#156	created_at,
#157	updated_at,
#158	is_deleted,
#159	actor_id,
#160	role,
#161	),
#162	)
#163	self.connection.execute("COMMIT")
#164	except Exception as e:
#165	self.connection.execute("ROLLBACK")
#166	logger.error(f"Failed to add history record: {e}")
#167	raise
#168
#169	def get_history(self, memory_id: str) -> List[Dict[str, Any]]:
#170	with self._lock:
#171	cur = self.connection.execute(
#172	"""
#173	SELECT id, memory_id, old_memory, new_memory, event,
#174	created_at, updated_at, is_deleted, actor_id, role
#175	FROM history
#176	WHERE memory_id = ?
#177	ORDER BY created_at ASC, DATETIME(updated_at) ASC
#178	""",
#179	(memory_id,),
#180	)
#181	rows = cur.fetchall()
#182
#183	return [
#184	{
#185	"id": r[0],
#186	"memory_id": r[1],
#187	"old_memory": r[2],
#188	"new_memory": r[3],
#189	"event": r[4],
#190	"created_at": r[5],
#191	"updated_at": r[6],
#192	"is_deleted": bool(r[7]),
#193	"actor_id": r[8],
#194	"role": r[9],
#195	}
#196	for r in rows
#197	]
#198
#199	def reset(self) -> None:
#200	"""Drop and recreate the history table."""
#201	with self._lock:
#202	try:
#203	self.connection.execute("BEGIN")
#204	self.connection.execute("DROP TABLE IF EXISTS history")
#205	self.connection.execute("COMMIT")
#206	self._create_history_table()
#207	except Exception as e:
#208	self.connection.execute("ROLLBACK")
#209	logger.error(f"Failed to reset history table: {e}")
#210	raise
#211
#212	def close(self) -> None:
#213	if self.connection:
#214	self.connection.close()
#215	self.connection = None
#216
#217	def __del__(self):
#218	self.close()
#219

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public