Update `GoogleDriveService` and add example usage notebook

.env.template

@@ -30,6 +30,7 @@ TEMPERATURE=0.8
 SYSTEM_PROMPT="You are a helpful teaching assistant for a data science class.\nBased on the students question, you will be given context retreived from class transcripts and materials to answer their question.\nYour responses should be:\n\n1. Accurate and based on the class content\n2. Clear and educational\n3. Concise but complete\nIf you're unsure about something, acknowledge it and suggest asking the professor."
 
 # Google Drive Configuration
+GOOGLE_DRIVE_ROOT_ID=1NB91EcIUXbOVcdCkXOAHdmWrDfgoh9fQ
 GOOGLE_PROJECT_ID=insufferable-slacker-123456
 GOOGLE_PRIVATE_KEY_ID=1a2b3c4d5e6f748891091d21304e506674829507
 GOOGLE_PRIVATE_KEY="-----BEGIN PRIVATE KEY-----\nMIIEvQIBADANBgkqhkiG9w0BAQEFAASC...\n-----END PRIVATE KEY-----\n"

pyproject.toml

@@ -21,6 +21,7 @@ classifiers = [
 dependencies = [
     "pydantic>=2.11.2",
     "pydantic-settings>=2.8.1",
+    "cachetools>=5.5.2",
     "more-itertools>=10.6.0",
     "python-dotenv>=1.1.0",
     "loguru>=0.7.3",

src/ctp_slack_bot/containers.py

@@ -11,6 +11,7 @@ from ctp_slack_bot.services.content_ingestion_service import ContentIngestionSer
 from ctp_slack_bot.services.context_retrieval_service import ContextRetrievalService
 from ctp_slack_bot.services.embeddings_model_service import EmbeddingsModelService
 from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
+from ctp_slack_bot.services.google_drive_service import GoogleDriveService
 from ctp_slack_bot.services.language_model_service import LanguageModelService
 from ctp_slack_bot.services.question_dispatch_service import QuestionDispatchService
 from ctp_slack_bot.services.schedule_service import ScheduleServiceResource
@@ -36,3 +37,4 @@ class Container(DeclarativeContainer):
     slack_bolt_app = Singleton(AsyncApp, token=settings.provided.SLACK_BOT_TOKEN().get_secret_value())
     slack_service = Resource(SlackServiceResource, event_brokerage_service=event_brokerage_service, slack_bolt_app=slack_bolt_app)
     socket_mode_handler = Singleton(lambda _, app, app_token: AsyncSocketModeHandler(app, app_token), slack_service, slack_bolt_app, settings.provided.SLACK_APP_TOKEN().get_secret_value())
+    google_drive_service = Singleton(GoogleDriveService, settings=settings)

src/ctp_slack_bot/core/config.py

@@ -49,6 +49,7 @@ class Settings(BaseSettings):
     SYSTEM_PROMPT: str
 
     # Google Drive Configuration
+    GOOGLE_DRIVE_ROOT_ID: str
     GOOGLE_PROJECT_ID: str
     GOOGLE_PRIVATE_KEY_ID: SecretStr
     GOOGLE_PRIVATE_KEY: SecretStr

src/ctp_slack_bot/models/__init__.py

@@ -1,3 +1,4 @@
 from ctp_slack_bot.models.base import Chunk, Content, VectorizedChunk, VectorQuery
+from ctp_slack_bot.models.google_drive import GoogleDriveMetadata
 from ctp_slack_bot.models.slack import SlackEventPayload, SlackMessage, SlackReaction, SlackResponse, SlackUserTimestampPair
 from ctp_slack_bot.models.webvtt import WebVTTContent, WebVTTFrame

src/ctp_slack_bot/models/google_drive.py

@@ -18,8 +18,8 @@ class GoogleDriveMetadata(BaseModel):
 
     @classmethod
     def from_folder_path_and_dict(cls: type["GoogleDriveMetadata"], folder_path: str, dict: dict) -> Self:
-        id = result["id"]
-        name = result["name"]
-        modified_time = datetime.fromisoformat(result["modifiedTime"])
-        mime_type = result["mimeType"]
+        id = dict["id"]
+        name = dict["name"]
+        modified_time = datetime.fromisoformat(dict["modifiedTime"])
+        mime_type = dict["mimeType"]
         return GoogleDriveMetadata(id=id, name=name, modified_time=modified_time, mime_type=mime_type, folder_path=folder_path)

src/ctp_slack_bot/services/__init__.py

@@ -3,6 +3,7 @@ from ctp_slack_bot.services.content_ingestion_service import ContentIngestionSer
 from ctp_slack_bot.services.context_retrieval_service import ContextRetrievalService
 from ctp_slack_bot.services.embeddings_model_service import EmbeddingsModelService
 from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
+from ctp_slack_bot.services.google_drive_service import GoogleDriveService
 from ctp_slack_bot.services.language_model_service import LanguageModelService
 from ctp_slack_bot.services.question_dispatch_service import QuestionDispatchService
 from ctp_slack_bot.services.slack_service import SlackService

src/ctp_slack_bot/services/google_drive_service.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from cachetools import TTLCache
 from google.oauth2 import service_account
 from googleapiclient.discovery import build
 from googleapiclient.http import MediaIoBaseDownload
@@ -9,16 +10,19 @@ from pydantic import BaseModel, PrivateAttr
 from typing import Collection, Dict, List, Optional, Self
 
 from ctp_slack_bot.core import Settings
+from ctp_slack_bot.models import GoogleDriveMetadata
+
 
 FOLDER_MIME_TYPE: str = "application/vnd.google-apps.folder"
-ROOT_FOLDER_NAME: str = "root"
+PATH_SEPARATOR: str = "/"
+
 
 class GoogleDriveService(BaseModel):
     """Service for interacting with Google Drive."""
 
     settings: Settings
     _google_drive_client: PrivateAttr = PrivateAttr()
-    _folder_cache: PrivateAttr = PrivateAttr(default_factory=dict)
+    _folder_cache: PrivateAttr = PrivateAttr(default_factory=lambda: TTLCache(maxsize=256, ttl=60))
 
     class Config:
         frozen=True
@@ -28,12 +32,12 @@ class GoogleDriveService(BaseModel):
         credentials = service_account.Credentials.from_service_account_info({
             "type": "service_account",
             "project_id": self.settings.GOOGLE_PROJECT_ID,
-            "private_key_id": self.settings.GOOGLE_PRIVATE_KEY_ID,
-            "private_key": self.settings.GOOGLE_PRIVATE_KEY,
+            "private_key_id": self.settings.GOOGLE_PRIVATE_KEY_ID.get_secret_value(),
+            "private_key": self.settings.GOOGLE_PRIVATE_KEY.get_secret_value(),
             "client_email": self.settings.GOOGLE_CLIENT_EMAIL,
             "client_id": self.settings.GOOGLE_CLIENT_ID,
             "token_uri": self.settings.GOOGLE_TOKEN_URI,
-        }, scopes=['https://www.googleapis.com/auth/drive'])
+        }, scopes=["https://www.googleapis.com/auth/drive"])
         self._google_drive_client = build('drive', 'v3', credentials=credentials)
         logger.debug("Created {}", self.__class__.__name__)
 
@@ -41,98 +45,92 @@ class GoogleDriveService(BaseModel):
         """Resolve a folder path to a Google Drive ID."""
 
         if not folder_path:
-            return ROOT_FOLDER_NAME
+            return self.settings.GOOGLE_DRIVE_ROOT_ID
             
         if folder_path in self._folder_cache:
             return self._folder_cache[folder_path]
 
-        current_id = ROOT_FOLDER_NAME
-        for part in folder_path.split('/'):
-            if not part:
-                continue
-            try:
+        current_id = self.settings.GOOGLE_DRIVE_ROOT_ID
+        try:
+            for part in folder_path.split(PATH_SEPARATOR):
                 results = self._google_drive_client.files().list(
-                    q=f"name='{part}' and mimeType='{FOLDER_MIME_TYPE}' and '{current_id}' in parents",
-                    fields='files(id,name)',
+                    q=f"name='{part.replace("\\", "\\\\").replace("'", "\\'")}' and mimeType='{FOLDER_MIME_TYPE}' and '{current_id}' in parents",
+                    fields="files(id,name)",
                     supportsAllDrives=True,
                     includeItemsFromAllDrives=True
                 ).execute()
-                if not results.get('files'):
-                    return None
-                current_id = results['files'][0]['id']
-            except HttpError as e:
-                logger.error("Error resolving folder path: {}", folder_path)
-                return None
+                match results:
+                    case {"files": [ {"id": id} ]}:
+                        current_id = id
+                    case _:
+                        logger.debug("Folder not found by path: {}", folder_path)
+                        return None
+        except HttpError as e:
+            logger.error("Error resolving folder path: {}", folder_path)
+            return None
 
         self._folder_cache[folder_path] = current_id
         return current_id
 
-    def list_directory(self: Self, folder_path: str) -> List[Dict]:
+    def list_directory(self: Self, folder_path: str) -> Collection[GoogleDriveMetadata]:
         """List contents of a directory with basic metadata."""
 
         folder_id = self._resolve_folder_id(folder_path)
         if not folder_id:
-            return []
+            logger.debug("Folder not found by path: {}", folder_path)
+            return ()
+
         try:
             results = self._google_drive_client.files().list(
                 q=f"'{folder_id}' in parents",
-                fields='files(id,name,mimeType,modifiedTime)',
+                fields="files(id,name,mimeType,modifiedTime)",
                 supportsAllDrives=True,
                 includeItemsFromAllDrives=True,
                 pageSize=1000
             ).execute()
-            
-            return [{
-                'id': f['id'],
-                'name': f['name'],
-                'modified': f['modifiedTime'],
-                'mime_type': f['mimeType']
-            } for f in results.get('files', [])]
+            return tuple(GoogleDriveMetadata.from_folder_path_and_dict(folder_path, result)
+                         for result
+                         in results.get('files', ()))
         except HttpError as e:
             logger.error("Error listing folder by path, {}: {}", folder_path, e)
-            return []
+            return ()
 
-    def get_metadata(self: Self, item_path: str) -> Optional[Dict]:
+    def get_metadata(self: Self, item_path: str) -> Optional[GoogleDriveMetadata]:
         """Get metadata for a specific file/folder by path."""
 
-        if '/' not in item_path:
-            folder_id = ROOT_FOLDER_NAME
-            item_name = item_path
-        else:
-            parts = item_path.split('/')
-            item_name = parts[-1]
-            folder_path = '/'.join(parts[:-1])
-            folder_id = self._resolve_folder_id(folder_path)
-            
+        match item_path.rsplit(PATH_SEPARATOR, 1):
+            case [item_name]:
+                folder_path = ""
+                folder_id = self.settings.GOOGLE_DRIVE_ROOT_ID
+            case [folder_path, item_name]:
+                folder_id = self._resolve_folder_id(folder_path)
+
         if not folder_id:
+            logger.debug("Folder not found by path: {}", folder_path)
             return None
             
         try:
             results = self._google_drive_client.files().list(
                 q=f"name='{item_name}' and '{folder_id}' in parents",
-                fields='files(id,name,mimeType,modifiedTime)',
+                fields="files(id,name,mimeType,modifiedTime)",
                 supportsAllDrives=True,
                 includeItemsFromAllDrives=True,
                 pageSize=1
             ).execute()
-            
-            if files := results.get('files'):
-                return {
-                    'id': files[0]['id'],
-                    'name': files[0]['name'],
-                    'modified': files[0]['modifiedTime'],
-                    'mime_type': files[0]['mimeType']
-                }
+            match results:
+                case {"files": [result]}:
+                    return GoogleDriveMetadata.from_folder_path_and_dict(folder_path, result)
         except HttpError as e:
             logger.error("Error getting metadata for item by path, {}: {}", item_path, e)
 
+        logger.debug("Item not found by path: {}", item_path)
         return None
 
     def read_file_by_id(self: Self, file_id: str) -> Optional[bytes]:
         """Read contents of a file by its unique identifier."""
 
         try:
-            request = self.service.files().get_media(fileId=file_id)
+            request = self._google_drive_client.files().get_media(fileId=file_id)
             buffer = BytesIO()
             downloader = MediaIoBaseDownload(buffer, request)
             done = False