Rename database.py to codette_memory_sql.py

codette_memory_sql.py

@@ -0,0 +1,92 @@
+import os
+import logging
+import pyodbc
+import sqlite3
+from typing import Optional, List, Tuple
+
+
+def connect_to_memory_database() -> Optional[pyodbc.Connection]:
+    """Connect to Azure SQL Database using HF secret environment variables."""
+    try:
+        server = os.getenv("AZURE_SQL_SERVER")
+        database = os.getenv("AZURE_SQL_DATABASE")
+        username = os.getenv("AZURE_SQL_USERNAME")
+        password = os.getenv("AZURE_SQL_PASSWORD")
+        driver = "{ODBC Driver 17 for SQL Server}"
+
+        if not all([server, database, username, password]):
+            logging.warning("Azure SQL credentials missing. Falling back to SQLite.")
+            return None
+
+        conn_str = (
+            f"DRIVER={driver};SERVER={server};DATABASE={database};"
+            f"UID={username};PWD={password}"
+        )
+        conn = pyodbc.connect(conn_str)
+        logging.info("Connected to Azure SQL Database.")
+        return conn
+    except pyodbc.Error as e:
+        logging.error(f"Azure SQL connection error: {e}")
+        return None
+
+
+def log_memory(conn, query: str, response: str) -> None:
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            "CREATE TABLE IF NOT EXISTS codette_memory (query TEXT, response TEXT)"
+        )
+        cursor.execute(
+            "INSERT INTO codette_memory (query, response) VALUES (?, ?)", (query, response)
+        )
+        conn.commit()
+        logging.info("Memory logged successfully.")
+    except Exception as e:
+        logging.error(f"Failed to log memory: {e}")
+
+
+def get_recent_entries(conn, limit: int = 5) -> List[Tuple[str, str]]:
+    try:
+        cursor = conn.cursor()
+        cursor.execute(
+            "SELECT TOP (?) query, response FROM codette_memory ORDER BY 1 DESC", (limit,)
+        )
+        return cursor.fetchall()
+    except Exception as e:
+        logging.warning(f"Failed to fetch memory: {e}")
+        return []
+
+
+def ping_database(conn) -> bool:
+    try:
+        cursor = conn.cursor()
+        cursor.execute("SELECT 1")
+        return True
+    except Exception as e:
+        logging.warning(f"Ping failed: {e}")
+        return False
+
+
+# SQLite fallback (in-memory or file-backed)
+def get_sqlite_memory(fallback_path: Optional[str] = ":memory:") -> sqlite3.Connection:
+    conn = sqlite3.connect(fallback_path)
+    conn.execute("CREATE TABLE IF NOT EXISTS codette_memory (query TEXT, response TEXT)")
+    return conn
+
+
+# Context manager for auto-closing
+class MemoryConnection:
+    def __init__(self):
+        self.conn = connect_to_memory_database()
+        if self.conn is None:
+            self.conn = get_sqlite_memory()
+
+    def __enter__(self):
+        return self.conn
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        try:
+            self.conn.close()
+            logging.info("Memory DB connection closed.")
+        except Exception as e:
+            logging.warning(f"Failed to close DB: {e}")

database.py

@@ -1,49 +0,0 @@
-import pyodbc
-import logging
-import os
-from dotenv import load_dotenv
-
-# Load environment variables from .env file
-load_dotenv()
-
-def connect_to_database() -> pyodbc.Connection:
-    """Connect to the Azure SQL Database."""
-    try:
-        server = os.getenv('AZURE_SQL_SERVER')
-        database = os.getenv('AZURE_SQL_DATABASE')
-        username = os.getenv('AZURE_SQL_USERNAME')
-        password = os.getenv('AZURE_SQL_PASSWORD')
-        driver = '{ODBC Driver 17 for SQL Server}'
-
-        conn_str = f'DRIVER={driver};SERVER={server};DATABASE={database};UID={username};PWD={password}'
-        conn = pyodbc.connect(conn_str)
-        logging.info(f"Connected to the Azure SQL Database at {server}")
-        return conn
-    except pyodbc.Error as e:
-        logging.error(f"Error connecting to the Azure SQL Database: {e}")
-        return None
-
-def close_database_connection(conn: pyodbc.Connection) -> None:
-    """Close the database connection."""
-    try:
-        if conn:
-            conn.close()
-            logging.info("Database connection closed.")
-    except pyodbc.Error as e:
-        logging.error(f"Error closing the database connection: {e}")
-
-# Example usage with context management
-class DatabaseConnection:
-    def __enter__(self) -> pyodbc.Connection:
-        self.conn = connect_to_database()
-        return self.conn
-
-    def __exit__(self, exc_type, exc_value, traceback) -> None:
-        close_database_connection(self.conn)
-
-# Example usage
-if __name__ == "__main__":
-    with DatabaseConnection() as conn:
-        if conn:
-            # Perform database operations
-            pass