Merge pull request #1123 from pipecat-ai/cb/sqlite

Add SQLite storage to the Gemini persistent storage example

examples/foundational/28c-transcription-processor-gemini.py

@@ -6,6 +6,7 @@
 
 import asyncio
 import os
+import sqlite3
 import sys
 from typing import List, Optional
 
@@ -44,22 +45,33 @@ class TranscriptHandler:
         output_file: Optional path to file where transcript is saved. If None, outputs to log only.
     """
 
-    def __init__(self, output_file: Optional[str] = None):
-        """Initialize handler with optional file output.
+    def __init__(self, output_file: Optional[str] = None, output_db: Optional[str] = None):
+        """Initialize handler with optional file or database output.
 
         Args:
             output_file: Path to output file. If None, outputs to log only.
         """
         self.messages: List[TranscriptionMessage] = []
         self.output_file: Optional[str] = output_file
+        self.output_db: Optional[str] = output_db
+
+        if self.output_db:
+            self.con = sqlite3.connect("example.db")
+            self.db = self.con.cursor()
+
+            table = self.db.execute("SELECT name FROM sqlite_master WHERE name='messages'")
+            if not (table.fetchone()):
+                self.db.execute(
+                    "CREATE TABLE messages(role TEXT, content TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP )"
+                )
         logger.debug(
-            f"TranscriptHandler initialized {'with output_file=' + output_file if output_file else 'with log output only'}"
+            f"TranscriptHandler initialized; output file: {output_file}, output DB: {output_db}"
         )
 
     async def save_message(self, message: TranscriptionMessage):
         """Save a single transcript message.
 
-        Outputs the message to the log and optionally to a file.
+        Outputs the message to the log and optionally to a SQLite database or file.
 
         Args:
             message: The message to save
@@ -78,6 +90,14 @@ class TranscriptHandler:
             except Exception as e:
                 logger.error(f"Error saving transcript message to file: {e}")
 
+        # and/or to a SQLite database
+        if self.output_db:
+            self.db.execute(
+                "INSERT INTO messages VALUES (?, ?, ?)",
+                (message.role, message.content, message.timestamp),
+            )
+            self.con.commit()
+
     async def on_transcript_update(
         self, processor: TranscriptProcessor, frame: TranscriptionUpdateFrame
     ):
@@ -136,7 +156,7 @@ async def main():
 
         # Create transcript processor and handler
         transcript = TranscriptProcessor()
-        transcript_handler = TranscriptHandler()  # Output to log only
+        transcript_handler = TranscriptHandler(output_db="example.db")  # Output to log only
         # transcript_handler = TranscriptHandler(output_file="transcript.txt") # Output to file and log
 
         pipeline = Pipeline(