Clean up and restore ability to shut down gracefully

pyproject.toml

@@ -24,7 +24,6 @@ dependencies = [
     "more-itertools>=10.6.0",
     "python-dotenv>=1.1.0",
     "loguru>=0.7.3",
-    "fastapi>=0.115.12",
     "dependency-injector>=4.46.0",
     "pytz>=2025.2",
     "apscheduler>=3.11.0",
@@ -36,7 +35,7 @@ dependencies = [
     "slack_bolt>=1.23.0",
     "pymongo>=4.11.3 ",
     "motor>=3.7.0",
-    "openai>=1.70.0"
+    "openai>=1.70.0",
     "google-api-python-client>=2.167.0",
     "google-auth>=2.39.0",
     "google-auth-oauthlib>=1.2.1"
@@ -50,7 +49,7 @@ dev = [
     "types-pytz>=2025.2",
     "black>=25.1.0",
     "isort>=6.0.1",
-    "ruff>=0.11.4",
+    "ruff>=0.11.4"
 ]
 
 [project.urls]

src/ctp_slack_bot/app.py

@@ -1,9 +1,24 @@
-from asyncio import run
+from asyncio import all_tasks, CancelledError, create_task, current_task, get_running_loop, run
 from loguru import logger
+from signal import SIGINT, SIGTERM
+from typing import Any, Callable
 
 from ctp_slack_bot.containers import Container
 from ctp_slack_bot.core.logging import setup_logging
 
+async def handle_shutdown_signal() -> None:
+    logger.info("Received shutdown signal.")
+    for task in all_tasks():
+        if task is not current_task() and not task.done():
+            task.cancel()
+            logger.trace("Cancelled task {}.", task.get_name())
+    logger.info("Cancelled all tasks.")
+
+def create_shutdown_signal_handler() -> Callable[[], None]:
+    def shutdown_signal_handler() -> None:
+        create_task(handle_shutdown_signal())
+    return shutdown_signal_handler
+
 async def main() -> None:
     # Setup logging.
     setup_logging()
@@ -16,19 +31,23 @@ async def main() -> None:
     # Kick off services which should be active from the start.
     container.content_ingestion_service()
     container.question_dispatch_service()
+    container.schedule_service()
 
-    # Start the scheduler.
-    schedule_service = container.schedule_service()
-    schedule_service.start()
-
-    # Start the Slack socket mode handler in a background thread.
+    # Start the Slack socket mode handler in the background.
     socket_mode_handler = container.socket_mode_handler()
-    logger.info("Starting Slack Socket Mode handler…")
-    await socket_mode_handler.start_async()
-
-    # Shutdown. (This will never execute, because the socket mode handler never returns.)
-    logger.info("Shutting down application…")
-    schedule_service.stop()
+    slack_bolt_task = create_task(socket_mode_handler.start_async())
+    shutdown_signal_handler = create_shutdown_signal_handler()
+    loop = get_running_loop()
+    loop.add_signal_handler(SIGINT, shutdown_signal_handler)
+    loop.add_signal_handler(SIGTERM, shutdown_signal_handler)
+    try:
+        logger.info("Starting Slack Socket Mode handler…")
+        await slack_bolt_task
+    except CancelledError:
+        logger.info("Shutting down application…")
+    finally:
+        await socket_mode_handler.close_async()
+        await container.shutdown_resources()
 
 if __name__ == "__main__":
     run(main())

src/ctp_slack_bot/containers.py

@@ -13,7 +13,7 @@ from ctp_slack_bot.services.embeddings_model_service import EmbeddingsModelServi
 from ctp_slack_bot.services.event_brokerage_service import EventBrokerageService
 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 ScheduleService
+from ctp_slack_bot.services.schedule_service import ScheduleServiceResource
 from ctp_slack_bot.services.slack_service import SlackServiceResource
 from ctp_slack_bot.services.vector_database_service import VectorDatabaseService
 from ctp_slack_bot.services.vectorization_service import VectorizationService
@@ -22,7 +22,7 @@ from ctp_slack_bot.services.vectorization_service import VectorizationService
 class Container(DeclarativeContainer):
     settings = Singleton(Settings)
     event_brokerage_service = Singleton(EventBrokerageService)
-    schedule_service = Singleton(ScheduleService, settings=settings)
+    schedule_service = Resource(ScheduleServiceResource, settings=settings)
     mongo_db = Resource(MongoDBResource, settings=settings) # TODO: generalize to any database.
     vectorized_chunk_repository = Singleton(MongoVectorizedChunkRepository, mongo_db=mongo_db)
     vector_database_service = Singleton(VectorDatabaseService, settings=settings, mongo_db=mongo_db)

src/ctp_slack_bot/core/logging.py

@@ -1,4 +1,4 @@
-from logging import __file__ as logging_file, basicConfig, currentframe, getLogger, Handler, INFO, LogRecord
+from logging import __file__ as logging_file, basicConfig, currentframe, getLogger, Handler, INFO, LogRecord, WARNING
 from loguru import logger
 from os import getenv
 from sys import stderr
@@ -90,7 +90,9 @@ def setup_logging() -> None:
     basicConfig(handlers=[InterceptHandler()], level=0, force=True)
 
     # Update logging levels for some noisy libraries.
-    for logger_name in ("uvicorn", "uvicorn.error", "fastapi", "httpx", "apscheduler", "pymongo"):
+    for logger_name in ("uvicorn", "uvicorn.error", "fastapi", "httpx", "pymongo"):
         getLogger(logger_name).setLevel(INFO)
+    for logger_name in ("apscheduler"):
+        getLogger(logger_name).setLevel(WARNING)
 
     logger.info(f"Logging configured with level {log_level}")

src/ctp_slack_bot/core/response_rendering.py

@@ -1,13 +0,0 @@
-from json import dumps
-from starlette.responses import JSONResponse
-from typing import Any, Self
-
-class PrettyJSONResponse(JSONResponse):
-    def render(self: Self, content: Any) -> bytes:
-        return dumps(
-            content,
-            ensure_ascii=False,
-            allow_nan=False,
-            indent=4,
-            separators=(", ", ": "),
-        ).encode()

src/ctp_slack_bot/db/mongo_db.py

@@ -1,13 +1,14 @@
-from dependency_injector.resources import Resource
+from asyncio import create_task
+from dependency_injector.resources import AsyncResource
 from motor.motor_asyncio import AsyncIOMotorClient
 from pymongo.errors import ConnectionFailure, ServerSelectionTimeoutError
 from pymongo.operations import SearchIndexModel
 from loguru import logger
 from pydantic import BaseModel, PrivateAttr
 from typing import Any, Dict, Optional, Self
-import asyncio
 
 from ctp_slack_bot.core.config import Settings
+from ctp_slack_bot.utils import sanitize_mongo_db_uri
 
 class MongoDB(BaseModel):
     """
@@ -16,23 +17,20 @@ class MongoDB(BaseModel):
     settings: Settings
     _client: PrivateAttr = PrivateAttr()
     _db: PrivateAttr = PrivateAttr()
-    
+
     class Config:
         arbitrary_types_allowed = True
-    
+
     def __init__(self: Self, **data: Dict[str, Any]) -> None:
         super().__init__(**data)
         logger.debug("Created {}", self.__class__.__name__)
-    
+
     def connect(self: Self) -> None:
         """Initialize MongoDB client with settings."""
         try:
             connection_string = self.settings.MONGODB_URI.get_secret_value()
-            logger.debug("Connecting to MongoDB using URI: {}", connection_string.replace(
-                connection_string.split('@')[-1].split('/')[0] if '@' in connection_string else '',
-                '[REDACTED]'
-            ))
-            
+            logger.debug("Connecting to MongoDB using URI: {}", sanitize_mongo_db_uri(connection_string))
+
             # Create client with appropriate settings
             self._client = AsyncIOMotorClient(
                 connection_string,
@@ -43,48 +41,48 @@ class MongoDB(BaseModel):
                 retryWrites=True,
                 w="majority"
             )
-            
+
             # Set database
             db_name = self.settings.MONGODB_NAME
-            
+
             self._db = self._client[db_name]
             logger.debug("MongoDB client initialized for database: {}", db_name)
-        
+
         except Exception as e:
             logger.error("Failed to initialize MongoDB client: {}", e)
             self._client = None
             self._db = None
             raise
-    
+
     @property
     def client(self: Self) -> AsyncIOMotorClient:
         """Get the MongoDB client instance."""
         if not hasattr(self, '_client') or self._client is None:
-            logger.warning("MongoDB client not initialized. Attempting to initialize.")
+            logger.warning("MongoDB client not initialized. Attempting to initialize…")
             self.connect()
             if not hasattr(self, '_client') or self._client is None:
-                raise ConnectionError("Failed to initialize MongoDB client")
+                raise ConnectionError("Failed to initialize MongoDB client.")
         return self._client
-    
+
     @property
     def db(self: Self) -> Any:
         """Get the MongoDB database instance."""
         if not hasattr(self, '_db') or self._db is None:
-            logger.warning("MongoDB database not initialized. Attempting to initialize client.")
+            logger.warning("MongoDB database not initialized. Attempting to initialize client…")
             self.connect()
             if not hasattr(self, '_db') or self._db is None:
-                raise ConnectionError("Failed to initialize MongoDB database")
+                raise ConnectionError("Failed to initialize MongoDB database.")
         return self._db
-    
+
     async def ping(self: Self) -> bool:
         """Check if MongoDB connection is alive."""
         try:
             # Get client to ensure we're connected
             client = self.client
-            
+
             # Try a simple ping command
             await client.admin.command('ping')
-            logger.debug("MongoDB connection is active")
+            logger.debug("MongoDB connection is active!")
             return True
         except (ConnectionFailure, ServerSelectionTimeoutError) as e:
             logger.error("MongoDB connection failed: {}", e)
@@ -92,7 +90,7 @@ class MongoDB(BaseModel):
         except Exception as e:
             logger.error("Unexpected error during MongoDB ping: {}", e)
             return False
-    
+
     async def get_collection(self: Self, name: str) -> Any:
         """
         Get a collection by name with validation.
@@ -100,29 +98,29 @@ class MongoDB(BaseModel):
         """
         # First ensure we can connect at all
         if not await self.ping():
-            logger.error("Cannot get collection '{}' - MongoDB connection is not available", name)
-            raise ConnectionError("MongoDB connection is not available")
-        
+            logger.error("Cannot get collection '{}' because a MongoDB connection is not available.", name)
+            raise ConnectionError("MongoDB connection is not available.")
+
         try:
             # Get all collection names to check if this one exists
-            logger.debug("Checking if collection '{}' exists", name)
+            logger.debug("Checking if collection '{}' exists…", name)
             collection_names = await self.db.list_collection_names()
             
             if name not in collection_names:
-                logger.info("Collection '{}' does not exist. Creating it.", name)
+                logger.info("Collection '{}' does not exist. Creating it…", name)
                 # Create the collection
                 await self.db.create_collection(name)
-                logger.debug("Successfully created collection '{}'", name)
+                logger.debug("Successfully created collection: {}", name)
             else:
-                logger.debug("Collection '{}' already exists", name)
-                
+                logger.debug("Collection '{}' already exists!", name)
+
             # Get and return the collection
             collection = self.db[name]
             return collection
         except Exception as e:
             logger.error("Error accessing collection '{}': {}", name, e)
             raise
-    
+
     async def create_indexes(self: Self, collection_name: str) -> None:
         """
         Create a vector search index on a collection.
@@ -131,7 +129,7 @@ class MongoDB(BaseModel):
             collection_name: Name of the collection
         """
         collection = await self.get_collection(collection_name)
-        
+
         try:
             # Create search index model using MongoDB's recommended approach
             search_index_model = SearchIndexModel(
@@ -149,41 +147,38 @@ class MongoDB(BaseModel):
                 name=f"{collection_name}_vector_index",
                 type="vectorSearch"
             )
-            
+
             # Create the search index using the motor collection
             result = await collection.create_search_index(search_index_model)
-            logger.info("Vector search index '{}' created for collection {}", result, collection_name)
-            
+            logger.info("Vector search index '{}' created for collection {}.", result, collection_name)
+
         except Exception as e:
             if "command not found" in str(e).lower():
                 logger.warning("Vector search not supported by this MongoDB instance. Some functionality may be limited.")
                 # Create a fallback standard index on embedding field
                 await collection.create_index("embedding")
-                logger.info("Created standard index on 'embedding' field as fallback")
+                logger.info("Created standard index on 'embedding' field as fallback.")
             else:
                 logger.error("Failed to create vector index: {}", e)
                 raise
-    
+
     async def close(self: Self) -> None:
         """Close MongoDB connection."""
         if self._client:
             self._client.close()
-            logger.info("MongoDB connection closed")
+            logger.info("Closed MongoDB connection.")
             self._client = None
             self._db = None
 
-class MongoDBResource(Resource):
-    def init(self: Self, settings: Settings) -> MongoDB:
+class MongoDBResource(AsyncResource):
+    async def init(self: Self, settings: Settings) -> MongoDB:
         logger.info("Initializing MongoDB connection for database: {}", settings.MONGODB_NAME)
         mongo_db = MongoDB(settings=settings)
         mongo_db.connect()
-        
-        # Test the connection asynchronously - this will run after init returns
-        asyncio.create_task(self._test_connection(mongo_db))
-        
+        await self._test_connection(mongo_db)
         return mongo_db
-    
-    async def _test_connection(self, mongo_db: MongoDB) -> None:
+
+    async def _test_connection(self: Self, mongo_db: MongoDB) -> None:
         """Test MongoDB connection and log the result."""
         try:
             is_connected = await mongo_db.ping()
@@ -193,11 +188,11 @@ class MongoDBResource(Resource):
                 logger.error("MongoDB connection test failed!")
         except Exception as e:
             logger.error("Error testing MongoDB connection: {}", e)
-    
+            raise
+
     async def shutdown(self: Self, mongo_db: MongoDB) -> None:
         """Close MongoDB connection on shutdown."""
         try:
-            logger.info("Closing MongoDB connection...")
             await mongo_db.close()
         except Exception as e:
             logger.error("Error closing MongoDB connection: {}", e)

src/ctp_slack_bot/services/GOOGLE_DRIVE_README.md

@@ -1,228 +0,0 @@
-# Google Drive Access Module
-
-This Python module provides a simplified way to interact with Google Drive, focusing on easy access to files in nested folders using path-like syntax. It handles various Google file formats and provides comprehensive metadata for files and folders.
-
-## Features
-
-- **Path-based folder access**: Access files using simple paths like `folder1/folder2/folder3`
-- **Efficient caching**: Folder IDs are cached to improve performance
-- **Comprehensive metadata**: Get detailed information about files and folders
-- **Read various file types**:
-  - Text files
-  - Google Docs
-  - VTT files
-- **Robust folder finding**: Works with exact and partial name matching
-- **Simple API**: Designed for ease of use with minimal code
-
-## Setup Instructions
-
-### 1. Create a Google Cloud Project
-
-1. Go to the [Google Cloud Console](https://console.cloud.google.com/)
-2. Click on the project dropdown at the top of the page and select "New Project"
-3. Enter a project name and click "Create"
-4. Once created, make sure your new project is selected in the dropdown
-
-### 2. Enable the Google Drive API
-
-1. In the Google Cloud Console, navigate to "APIs & Services" > "Library" in the left sidebar
-2. Search for "Google Drive API" in the search bar
-3. Click on "Google Drive API" in the results
-4. Click the "Enable" button
-
-### 3. Create OAuth Credentials
-
-1. In the Google Cloud Console, go to "APIs & Services" > "Credentials" in the left sidebar
-2. Click "Create Credentials" at the top and select "OAuth client ID"
-3. If prompted to configure the OAuth consent screen:
-   - Choose "External" user type (or "Internal" if you're in a Google Workspace organization)
-   - Fill in the required information (App name, User support email, Developer contact email)
-   - Click "Save and Continue"
-   - Add the following scopes:
-     - `.../auth/drive` (Full access to Google Drive)
-   - Click "Save and Continue" and complete the registration
-4. Return to the "Create OAuth client ID" screen
-5. Select "Desktop application" as the Application type
-6. Enter a name for your OAuth client (e.g., "Google Drive Access Desktop")
-7. Click "Create"
-8. Download the JSON file (this is your `client_secret.json`)
-
-### 4. Project Setup
-
-1. Setup a virtual environment and install dependencies:
-```bash
-python -m venv venv
-source venv/bin/activate  # On Windows: venv\Scripts\activate
-pip install -r requirements.txt
-```
-
-2. Place your credentials:
-   - Create a `credentials` directory in your project root
-   - Move the downloaded OAuth client JSON file to the `credentials` directory
-   - Rename it to `client_secret.json`
-
-### 5. Authentication Process
-
-When you run the application for the first time:
-1. A browser window will open automatically
-2. You'll be asked to sign in to your Google account
-3. You'll see a consent screen asking for permission to access your Google Drive
-4. After granting permission, the browser will display a success message
-5. The application will save a token file (`token.pickle`) in the credentials directory for future use
-
-## Usage Guide
-
-The `EasyGoogleDrive` class provides several methods to interact with Google Drive. Here's how to use the core functionality:
-
-### Basic Usage
-
-```python
-from google_drive_access import EasyGoogleDrive
-
-# Initialize the Google Drive client
-drive = EasyGoogleDrive()
-
-# Example folder path - replace with your actual folder path
-folder_path = "Spring-2025-BAI"
-subfolder_path = "Spring-2025-BAI/transcripts"
-```
-
-### Listing Folders
-
-```python
-# List folders in a directory
-folders = drive.get_folders_in_folder(folder_path)
-
-# Access folder properties
-for folder in folders:
-    print(f"Folder: {folder['name']}")
-    print(f"  Created: {folder.get('createdTimeFormatted', 'Unknown')}")
-    print(f"  Modified: {folder.get('modifiedTimeFormatted', 'Unknown')}")
-```
-
-### Listing Files
-
-```python
-# List files in a directory
-files = drive.get_files_in_folder(subfolder_path)
-
-# Access file properties
-for file in files:
-    print(f"File: {file['name']}")
-    print(f"  Type: {file.get('fileType', 'Unknown')}")
-    print(f"  Created: {file.get('createdTimeFormatted', 'Unknown')}")
-    print(f"  Modified: {file.get('modifiedTimeFormatted', 'Unknown')}")
-    print(f"  Size: {file.get('sizeFormatted', 'Unknown')}")
-```
-
-### Getting a Specific File
-
-```python
-# Get a specific file with metadata
-file = drive.get_file("example.txt", subfolder_path, include_metadata=True)
-
-if file:
-    print(f"File: {file['name']}")
-    print(f"  Type: {file.get('fileType', 'Unknown')}")
-    print(f"  Created: {file.get('createdTimeFormatted', 'Unknown')}")
-    print(f"  Modified: {file.get('modifiedTimeFormatted', 'Unknown')}")
-    print(f"  Size: {file.get('sizeFormatted', 'Unknown')}")
-```
-
-### Getting All Items in a Folder
-
-```python
-# Get all items (files and folders) in a folder
-all_items = drive.get_all_files_in_folder(folder_path)
-
-# Access item properties
-for item in all_items:
-    item_type = "Folder" if item.get('mimeType') == drive.MIME_TYPES['folder'] else item.get('fileType', 'Unknown')
-    print(f"Item: {item['name']} ({item_type})")
-```
-
-### Checking if a File Exists
-
-```python
-# Check if a file exists
-exists = drive.file_exists("example.txt", subfolder_path)
-print(f"File exists: {exists}")
-```
-
-### Getting File Modified Time
-
-```python
-# Get file modified time
-modified_time = drive.get_file_modified_time("example.txt", subfolder_path)
-if modified_time:
-    print(f"Last modified: {modified_time}")
-```
-
-### Reading File Content
-
-```python
-# Get file with content
-file_with_content = drive.get_file("example.txt", subfolder_path, include_content=True)
-
-if file_with_content and 'file_content' in file_with_content:
-    content = file_with_content['file_content']
-    if content:
-        print(f"Content: {content[:100]}...")  # Print first 100 characters
-```
-
-## Complete Example
-
-For a complete example of how to use the `EasyGoogleDrive` class, see the `basic_usage.py` file included in this package. This file demonstrates all the core functionality with practical examples.
-
-## Key Concepts
-
-### Path-based Folder Access
-
-The module uses a simple path-like syntax to access folders:
-
-```python
-# Access a deeply nested folder
-folder_path = "folder1/folder2/folder3"
-files = drive.get_files_in_folder(folder_path)
-```
-
-This makes it much easier to work with nested folder structures compared to using folder IDs.
-
-### Metadata Fields
-
-The module provides comprehensive metadata for files and folders, including:
-
-- **Creation and modification dates**: Both as datetime objects and formatted strings
-- **File size**: Both in bytes and human-readable format (KB, MB, GB)
-- **File type**: Simplified type based on MIME type
-- **Owner information**: Names and email addresses of file owners
-- **Sharing status**: Whether the file is shared
-- **Web links**: Direct links to view the file in a browser
-
-## Error Handling
-
-The module includes comprehensive error handling:
-
-- **Authentication errors**: Clear messages when credentials are missing or invalid
-- **Folder not found**: Helpful messages when a folder in the path cannot be found
-- **File not found**: Attempts partial name matching before giving up
-- **Decoding errors**: Handles issues with file content encoding
-
-## Dependencies
-
-- **Required**:
-  - google-auth-oauthlib
-  - google-auth-httplib2
-  - google-api-python-client
-  - python-dateutil
-
-## Security Notes
-
-- Never commit your `client_secret.json` or token files to version control
-- Add `credentials/` to your `.gitignore` file
-- Keep your credentials secure and don't share them
-- For production applications, consider using service accounts with the minimum required permissions
-
-## Contributing
-
-Feel free to contribute to this project by submitting issues or pull requests.

src/ctp_slack_bot/services/google_drive_access.py

@@ -1,623 +0,0 @@
-"""
-Easy Google Drive Access
-
-A simplified module for accessing Google Drive files in nested folders.
-Designed to make it as easy as possible to access files using path-like syntax.
-"""
-
-import os
-import pickle
-import io
-import datetime
-from typing import List, Dict, Optional, Any, Union
-
-from google.oauth2.credentials import Credentials
-from google_auth_oauthlib.flow import InstalledAppFlow
-from google.auth.transport.requests import Request
-from googleapiclient.discovery import build
-from googleapiclient.http import MediaIoBaseDownload
-from googleapiclient.errors import HttpError
-
-
-class EasyGoogleDrive:
-    """
-    Simplified Google Drive access focused on accessing files in nested folders.
-    """
-    
-    # Define the scopes needed for the application
-    SCOPES = ['https://www.googleapis.com/auth/drive']
-
-    # Define common MIME types
-    MIME_TYPES = {
-        'folder': 'application/vnd.google-apps.folder',
-        'document': 'application/vnd.google-apps.document',
-        'spreadsheet': 'application/vnd.google-apps.spreadsheet',
-        'text': 'text/plain',
-        'pdf': 'application/pdf',
-        'image': 'image/jpeg',
-        'video': 'video/mp4',
-        'audio': 'audio/mpeg',
-    }
-
-    # Define metadata fields to retrieve
-    FILE_FIELDS = 'id, name, mimeType, createdTime, modifiedTime, size, description, webViewLink, thumbnailLink, owners, shared, sharingUser, lastModifyingUser, capabilities, permissions'
-    FOLDER_FIELDS = 'id, name, createdTime, modifiedTime, description, webViewLink, owners, shared, sharingUser, lastModifyingUser, capabilities, permissions'
-
-    def __init__(self, credentials_dir: str = 'credentials'):
-        """Initialize the Google Drive access."""
-        self.credentials_dir = credentials_dir
-        self.credentials_path = os.path.join(credentials_dir, 'client_secret.json')
-        self.token_path = os.path.join(credentials_dir, 'token.pickle')
-        
-        # Ensure credentials directory exists
-        os.makedirs(credentials_dir, exist_ok=True)
-        
-        # Initialize the Drive API service
-        self.service = build('drive', 'v3', credentials=self._get_credentials())
-        
-        # Cache for folder IDs to avoid repeated lookups
-        self.folder_id_cache = {}
-
-    def _get_credentials(self) -> Credentials:
-        """Get and refresh Google Drive API credentials."""
-        creds = None
-
-        # Load existing token if it exists
-        if os.path.exists(self.token_path):
-            with open(self.token_path, 'rb') as token:
-                creds = pickle.load(token)
-
-        # If credentials need refresh or don't exist
-        if not creds or not creds.valid:
-            if creds and creds.expired and creds.refresh_token:
-                creds.refresh(Request())
-            else:
-                if not os.path.exists(self.credentials_path):
-                    raise FileNotFoundError(
-                        f"Client secrets file not found at {self.credentials_path}. "
-                        "Please follow the setup instructions in the README."
-                    )
-                
-                flow = InstalledAppFlow.from_client_secrets_file(
-                    self.credentials_path, self.SCOPES)
-                creds = flow.run_local_server(port=0)
-
-            # Save the credentials for future use
-            with open(self.token_path, 'wb') as token:
-                pickle.dump(creds, token)
-
-        return creds
-
-    def _format_metadata(self, metadata: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        Format metadata for easier reading and usage.
-        
-        Args:
-            metadata: Raw metadata from Google Drive API
-            
-        Returns:
-            Formatted metadata dictionary
-        """
-        formatted = metadata.copy()
-        
-        # Format dates
-        for date_field in ['createdTime', 'modifiedTime']:
-            if date_field in formatted:
-                try:
-                    # Convert ISO 8601 string to datetime object
-                    dt = datetime.datetime.fromisoformat(formatted[date_field].replace('Z', '+00:00'))
-                    formatted[date_field] = dt
-                    # Add a formatted date string for easier reading
-                    formatted[f"{date_field}Formatted"] = dt.strftime('%Y-%m-%d %H:%M:%S')
-                except (ValueError, AttributeError):
-                    pass
-        
-        # Format size
-        if 'size' in formatted and formatted['size']:
-            try:
-                size_bytes = int(formatted['size'])
-                # Add human-readable size
-                if size_bytes < 1024:
-                    formatted['sizeFormatted'] = f"{size_bytes} B"
-                elif size_bytes < 1024 * 1024:
-                    formatted['sizeFormatted'] = f"{size_bytes / 1024:.1f} KB"
-                elif size_bytes < 1024 * 1024 * 1024:
-                    formatted['sizeFormatted'] = f"{size_bytes / (1024 * 1024):.1f} MB"
-                else:
-                    formatted['sizeFormatted'] = f"{size_bytes / (1024 * 1024 * 1024):.1f} GB"
-            except (ValueError, TypeError):
-                pass
-                
-        # Extract owner names
-        if 'owners' in formatted and formatted['owners']:
-            formatted['ownerNames'] = [owner.get('displayName', 'Unknown') for owner in formatted['owners']]
-            formatted['ownerEmails'] = [owner.get('emailAddress', 'Unknown') for owner in formatted['owners']]
-            
-        # Add file type description
-        if 'mimeType' in formatted:
-            mime_type = formatted['mimeType']
-            for key, value in self.MIME_TYPES.items():
-                if mime_type == value:
-                    formatted['fileType'] = key
-                    break
-            else:
-                # If not found in our predefined types
-                formatted['fileType'] = mime_type.split('/')[-1]
-                
-        return formatted
-
-    def get_folder_id(self, folder_path: str) -> Optional[str]:
-        """
-        Get a folder ID from a path like 'folder1/folder2/folder3'.
-        
-        Args:
-            folder_path: Path to the folder, using '/' as separator
-            
-        Returns:
-            The folder ID if found, None otherwise
-        """
-        # Check if we've already resolved this path
-        if folder_path in self.folder_id_cache:
-            return self.folder_id_cache[folder_path]
-            
-        # If it looks like an ID already, return it
-        if len(folder_path) > 25 and '/' not in folder_path:
-            return folder_path
-            
-        # Split the path into components
-        parts = folder_path.split('/')
-        
-        # Start from the root
-        current_folder_id = None
-        current_path = ""
-        
-        # Traverse the path one folder at a time
-        for i, folder_name in enumerate(parts):
-            if not folder_name:  # Skip empty parts
-                continue
-                
-            # Update the current path for caching
-            if current_path:
-                current_path += f"/{folder_name}"
-            else:
-                current_path = folder_name
-                
-            # Check if we've already resolved this subpath
-            if current_path in self.folder_id_cache:
-                current_folder_id = self.folder_id_cache[current_path]
-                continue
-                
-            # Search for the folder by name
-            query = f"mimeType='{self.MIME_TYPES['folder']}' and name='{folder_name}'"
-            if current_folder_id:
-                query += f" and '{current_folder_id}' in parents"
-                
-            try:
-                results = self.service.files().list(
-                    q=query,
-                    spaces='drive',
-                    fields='files(id, name)',
-                    pageSize=10
-                ).execute()
-                
-                files = results.get('files', [])
-                if not files:
-                    # Try a more flexible search if exact match fails
-                    query = query.replace(f"name='{folder_name}'", f"name contains '{folder_name}'")
-                    results = self.service.files().list(
-                        q=query,
-                        spaces='drive',
-                        fields='files(id, name)',
-                        pageSize=10
-                    ).execute()
-                    
-                    files = results.get('files', [])
-                    if not files:
-                        print(f"Could not find folder '{folder_name}' in path '{folder_path}'")
-                        return None
-                        
-                # Use the first match
-                current_folder_id = files[0]['id']
-                
-                # Cache this result
-                self.folder_id_cache[current_path] = current_folder_id
-                
-            except HttpError as error:
-                print(f"Error finding folder: {error}")
-                return None
-                
-        return current_folder_id
-
-    def get_folders_in_folder(self, folder_path: str, include_metadata: bool = True) -> List[Dict[str, Any]]:
-        """
-        Get all subfolders in a folder specified by path.
-        
-        Args:
-            folder_path: Path to the folder, using '/' as separator
-            include_metadata: Whether to include detailed metadata (default: True)
-            
-        Returns:
-            List of folder metadata dictionaries
-        """
-        # Get the folder ID
-        folder_id = self.get_folder_id(folder_path)
-        if not folder_id:
-            print(f"Could not find folder: '{folder_path}'")
-            return []
-            
-        # List all folders in this folder
-        query = f"'{folder_id}' in parents and mimeType = '{self.MIME_TYPES['folder']}'"
-        
-        try:
-            results = self.service.files().list(
-                q=query,
-                spaces='drive',
-                fields=f'files({self.FOLDER_FIELDS})' if include_metadata else 'files(id, name)',
-                pageSize=1000
-            ).execute()
-            
-            folders = results.get('files', [])
-            
-            # Format metadata if requested
-            if include_metadata and folders:
-                folders = [self._format_metadata(folder) for folder in folders]
-            
-            if folders:
-                print(f"Found {len(folders)} subfolders in '{folder_path}':")
-                for folder in folders:
-                    if include_metadata and 'createdTimeFormatted' in folder:
-                        print(f"  - {folder['name']} (Created: {folder['createdTimeFormatted']})")
-                    else:
-                        print(f"  - {folder['name']}")
-            else:
-                print(f"No subfolders found in '{folder_path}'")
-                
-            return folders
-            
-        except HttpError as error:
-            print(f"Error listing folders: {error}")
-            return []
-
-    def get_files_in_folder(self, folder_path: str, include_metadata: bool = True, include_content: bool = False) -> List[Dict[str, Any]]:
-        """
-        Get all files in a folder specified by path.
-        
-        Args:
-            folder_path: Path to the folder, using '/' as separator
-            include_metadata: Whether to include detailed metadata (default: True)
-            include_content: Whether to include file content (default: False)
-            
-        Returns:
-            List of file metadata dictionaries, optionally including file content
-        """
-        # Get the folder ID
-        folder_id = self.get_folder_id(folder_path)
-        if not folder_id:
-            print(f"Could not find folder: '{folder_path}'")
-            return []
-            
-        # List all non-folder files in this folder
-        query = f"'{folder_id}' in parents and mimeType != '{self.MIME_TYPES['folder']}'"
-        
-        try:
-            results = self.service.files().list(
-                q=query,
-                spaces='drive',
-                fields=f'files({self.FILE_FIELDS})' if include_metadata else 'files(id, name, mimeType)',
-                pageSize=1000
-            ).execute()
-            
-            files = results.get('files', [])
-            
-            # Format metadata if requested
-            if include_metadata and files:
-                files = [self._format_metadata(file) for file in files]
-            
-            # Add file content if requested
-            if include_content and files:
-                for file in files:
-                    try:
-                        # Skip files that are likely not text-based
-                        if any(ext in file['name'].lower() for ext in ['.jpg', '.png', '.gif', '.mp3', '.mp4']):
-                            print(f"Skipping content for non-text file: {file['name']}")
-                            file['file_content'] = None
-                            continue
-                            
-                        # Read the file content
-                        content = self.read_file_from_object(file)
-                        file['file_content'] = content
-                        
-                        if content is not None:
-                            print(f"Successfully read content for: {file['name']} ({len(content)} characters)")
-                        else:
-                            print(f"Unable to read content for: {file['name']}")
-                    except Exception as e:
-                        print(f"Error reading content for {file['name']}: {e}")
-                        file['file_content'] = None
-            
-            if files:
-                print(f"Found {len(files)} files in '{folder_path}':")
-                for file in files:
-                    if include_metadata and 'createdTimeFormatted' in file:
-                        print(f"  - {file['name']} ({file.get('fileType', 'Unknown')}, Created: {file['createdTimeFormatted']})")
-                    else:
-                        print(f"  - {file['name']} ({file.get('mimeType', 'Unknown')})")
-            else:
-                print(f"No files found in '{folder_path}'")
-                
-            return files
-            
-        except HttpError as error:
-            print(f"Error listing files: {error}")
-            return []
-
-    def get_file(self, file_name: str, folder_path: str, include_metadata: bool = True, include_content: bool = False) -> Optional[Dict[str, Any]]:
-        """
-        Get a specific file by name from a folder.
-        
-        Args:
-            file_name: Name of the file to get
-            folder_path: Path to the folder containing the file
-            include_metadata: Whether to include detailed metadata (default: True)
-            include_content: Whether to include file content (default: False)
-            
-        Returns:
-            File metadata dictionary, optionally including content, or None if file not found
-        """
-        # Get the folder ID
-        folder_id = self.get_folder_id(folder_path)
-        if not folder_id:
-            print(f"Could not find folder: '{folder_path}'")
-            return None
-            
-        # Find the file by name in this folder
-        query = f"'{folder_id}' in parents and name = '{file_name}'"
-        
-        try:
-            results = self.service.files().list(
-                q=query,
-                spaces='drive',
-                fields=f'files({self.FILE_FIELDS})' if include_metadata else 'files(id, name, mimeType)',
-                pageSize=1
-            ).execute()
-            
-            files = results.get('files', [])
-            if not files:
-                # Try a more flexible search
-                query = query.replace(f"name = '{file_name}'", f"name contains '{file_name}'")
-                results = self.service.files().list(
-                    q=query,
-                    spaces='drive',
-                    fields=f'files({self.FILE_FIELDS})' if include_metadata else 'files(id, name, mimeType)',
-                    pageSize=10
-                ).execute()
-                
-                files = results.get('files', [])
-                if not files:
-                    print(f"Could not find file '{file_name}' in '{folder_path}'")
-                    return None
-                    
-            # Use the first match
-            file = files[0]
-            
-            # Format metadata if requested
-            if include_metadata:
-                file = self._format_metadata(file)
-            
-            # Add file content if requested
-            if include_content:
-                try:
-                    # Skip files that are likely not text-based
-                    if any(ext in file['name'].lower() for ext in ['.jpg', '.png', '.gif', '.mp3', '.mp4']):
-                        print(f"Skipping content for non-text file: {file['name']}")
-                        file['file_content'] = None
-                    else:
-                        # Read the file content
-                        content = self.read_file_from_object(file)
-                        file['file_content'] = content
-                        
-                        if content is not None:
-                            print(f"Successfully read content for: {file['name']} ({len(content)} characters)")
-                        else:
-                            print(f"Unable to read content for: {file['name']}")
-                except Exception as e:
-                    print(f"Error reading content for {file['name']}: {e}")
-                    file['file_content'] = None
-            
-            print(f"Found file: {file['name']}")
-            return file
-                
-        except HttpError as error:
-            print(f"Error getting file: {error}")
-            return None
-
-    def get_all_files_in_folder(self, folder_path: str, include_metadata: bool = True, include_content: bool = False) -> List[Dict[str, Any]]:
-        """
-        Get all items (files and folders) in a folder specified by path.
-        
-        Args:
-            folder_path: Path to the folder, using '/' as separator
-            include_metadata: Whether to include detailed metadata (default: True)
-            include_content: Whether to include file content (default: False)
-            
-        Returns:
-            List of file and folder metadata dictionaries, optionally including file content
-        """
-        # Get the folder ID
-        folder_id = self.get_folder_id(folder_path)
-        if not folder_id:
-            print(f"Could not find folder: '{folder_path}'")
-            return []
-            
-        # List all items in this folder
-        query = f"'{folder_id}' in parents"
-        
-        try:
-            results = self.service.files().list(
-                q=query,
-                spaces='drive',
-                fields=f'files({self.FILE_FIELDS})' if include_metadata else 'files(id, name, mimeType)',
-                pageSize=1000
-            ).execute()
-            
-            items = results.get('files', [])
-            
-            # Format metadata if requested
-            if include_metadata and items:
-                items = [self._format_metadata(item) for item in items]
-            
-            # Add file content if requested
-            if include_content and items:
-                for item in items:
-                    # Skip folders and non-text files
-                    if item.get('mimeType') == self.MIME_TYPES['folder'] or any(ext in item['name'].lower() for ext in ['.jpg', '.png', '.gif', '.mp3', '.mp4']):
-                        item['file_content'] = None
-                        continue
-                        
-                    try:
-                        # Read the file content
-                        content = self.read_file_from_object(item)
-                        item['file_content'] = content
-                        
-                        if content is not None:
-                            print(f"Successfully read content for: {item['name']} ({len(content)} characters)")
-                        else:
-                            print(f"Unable to read content for: {item['name']}")
-                    except Exception as e:
-                        print(f"Error reading content for {item['name']}: {e}")
-                        item['file_content'] = None
-            
-            if items:
-                print(f"Found {len(items)} items in '{folder_path}':")
-                for item in items:
-                    if include_metadata and 'createdTimeFormatted' in item:
-                        item_type = 'Folder' if item.get('mimeType') == self.MIME_TYPES['folder'] else item.get('fileType', 'Unknown')
-                        print(f"  - {item['name']} ({item_type}, Created: {item['createdTimeFormatted']})")
-                    else:
-                        item_type = 'Folder' if item.get('mimeType') == self.MIME_TYPES['folder'] else item.get('mimeType', 'Unknown')
-                        print(f"  - {item['name']} ({item_type})")
-            else:
-                print(f"No items found in '{folder_path}'")
-                
-            return items
-            
-        except HttpError as error:
-            print(f"Error listing items: {error}")
-            return []
-
-    def file_exists(self, file_name: str, folder_path: str) -> bool:
-        """
-        Check if a file exists at the specified path in Google Drive.
-        
-        Args:
-            file_name: Name of the file to check
-            folder_path: Path to the folder containing the file
-            
-        Returns:
-            True if the file exists, False otherwise
-        """
-        # Get the folder ID
-        folder_id = self.get_folder_id(folder_path)
-        if not folder_id:
-            print(f"Could not find folder: '{folder_path}'")
-            return False
-            
-        # Check if the file exists in this folder
-        query = f"'{folder_id}' in parents and name = '{file_name}'"
-        
-        try:
-            results = self.service.files().list(
-                q=query,
-                spaces='drive',
-                fields='files(id, name)',
-                pageSize=1
-            ).execute()
-            
-            files = results.get('files', [])
-            if not files:
-                # Try a more flexible search
-                query = query.replace(f"name = '{file_name}'", f"name contains '{file_name}'")
-                results = self.service.files().list(
-                    q=query,
-                    spaces='drive',
-                    fields='files(id, name)',
-                    pageSize=10
-                ).execute()
-                
-                files = results.get('files', [])
-                if not files:
-                    print(f"File '{file_name}' does not exist in '{folder_path}'")
-                    return False
-                    
-            # File exists
-            print(f"File '{file_name}' exists in '{folder_path}'")
-            return True
-            
-        except HttpError as error:
-            print(f"Error checking if file exists: {error}")
-            return False
-
-    def get_file_modified_time(self, file_name: str, folder_path: str) -> Optional[datetime.datetime]:
-        """
-        Get the last modified time of a file.
-        
-        Args:
-            file_name: Name of the file
-            folder_path: Path to the folder containing the file
-            
-        Returns:
-            The last modified time as a datetime object, or None if the file doesn't exist
-        """
-        # Get the file metadata
-        file = self.get_file(file_name, folder_path, include_metadata=True)
-        if not file:
-            return None
-            
-        # Return the modified time
-        return file.get('modifiedTime')
-
-    def read_file_from_object(self, file_object: Dict[str, Any]) -> Optional[str]:
-        """
-        Read the contents of a file using a file object.
-        
-        Args:
-            file_object: A Google file object with at least 'id' and 'mimeType' fields
-            
-        Returns:
-            The file contents as a string, or None if the file couldn't be read
-        """
-        file_id = file_object.get('id')
-        mime_type = file_object.get('mimeType')
-        
-        if not file_id or not mime_type:
-            print("File object is missing 'id' or 'mimeType' fields.")
-            return None
-            
-        try:
-            # Read the file based on its type
-            if mime_type == self.MIME_TYPES['document']:
-                # Export Google Doc as plain text
-                response = self.service.files().export(
-                    fileId=file_id,
-                    mimeType='text/plain'
-                ).execute()
-                return response.decode('utf-8')
-                
-            else:
-                # Download regular files
-                request = self.service.files().get_media(fileId=file_id)
-                fh = io.BytesIO()
-                downloader = MediaIoBaseDownload(fh, request)
-                
-                done = False
-                while not done:
-                    _, done = downloader.next_chunk()
-                    
-                return fh.getvalue().decode('utf-8')
-                
-        except HttpError as error:
-            print(f"Error reading file: {error}")
-            return None
-        except Exception as e:
-            print(f"Error decoding file content: {e}")
-            return None 

src/ctp_slack_bot/services/google_drive_basic_usage.py

@@ -1,178 +0,0 @@
-"""
-Basic Usage Examples for EasyGoogleDrive
-
-This file demonstrates how to use the EasyGoogleDrive class to interact with Google Drive.
-It provides examples of the main functionality without printing all results to keep the output clean.
-"""
-
-from google_drive_access import EasyGoogleDrive
-import datetime
-
-def main():
-    """
-    Main function demonstrating the basic usage of EasyGoogleDrive.
-    """
-    # Initialize the Google Drive client
-    # This will prompt for authentication the first time it's run
-    drive = EasyGoogleDrive()
-    
-    # Example folder path - replace with your actual folder path
-    folder_path = "Spring-2025-BAI"
-    subfolder_path = "Spring-2025-BAI/transcripts"
-    
-    print("=== Basic Usage Examples for EasyGoogleDrive ===\n")
-    
-    # Example 1: List folders in a directory
-    print("Example 1: Listing folders in a directory")
-    print("----------------------------------------")
-    folders = drive.get_folders_in_folder(folder_path)
-    
-    # Print only the first 3 folders (if any exist)
-    if folders:
-        print(f"Found {len(folders)} folders. Showing first 3:")
-        for i, folder in enumerate(folders[:3]):
-            print(f"  - {folder['name']} (Created: {folder.get('createdTimeFormatted', 'Unknown')})")
-        if len(folders) > 3:
-            print(f"  ... and {len(folders) - 3} more folders")
-    else:
-        print("No folders found.")
-    print()
-    
-    # Example 2: List files in a directory
-    print("Example 2: Listing files in a directory")
-    print("--------------------------------------")
-    files = drive.get_files_in_folder(subfolder_path)
-    
-    # Print only the first 3 files (if any exist)
-    if files:
-        print(f"Found {len(files)} files. Showing first 3:")
-        for i, file in enumerate(files[:3]):
-            file_type = file.get('fileType', 'Unknown')
-            created_time = file.get('createdTimeFormatted', 'Unknown')
-            print(f"  - {file['name']} ({file_type}, Created: {created_time})")
-        if len(files) > 3:
-            print(f"  ... and {len(files) - 3} more files")
-    else:
-        print("No files found.")
-    print()
-    
-    # Example 3: Get a specific file
-    print("Example 3: Getting a specific file")
-    print("--------------------------------")
-    # Use the first file found in the previous example, or a default if none were found
-    file_name = files[-1]['name'] if files and len(files) > 0 else "example.txt"
-    
-    file = drive.get_file(file_name, subfolder_path, include_metadata=True)
-    if file:
-        print(f"File found: {file['name']}")
-        print(f"  Type: {file.get('fileType', 'Unknown')}")
-        print(f"  Created: {file.get('createdTimeFormatted', 'Unknown')}")
-        print(f"  Modified: {file.get('modifiedTimeFormatted', 'Unknown')}")
-        print(f"  Size: {file.get('sizeFormatted', 'Unknown')}")
-    else:
-        print(f"File '{file_name}' not found.")
-    print()
-    
-    # Example 4: Get all items in a folder (files and folders)
-    print("Example 4: Getting all items in a folder")
-    print("--------------------------------------")
-    all_items = drive.get_all_files_in_folder(folder_path)
-    
-    # Print only the first 3 items (if any exist)
-    if all_items:
-        print(f"Found {len(all_items)} items. Showing first 3:")
-        for i, item in enumerate(all_items[:3]):
-            item_type = "Folder" if item.get('mimeType') == drive.MIME_TYPES['folder'] else item.get('fileType', 'Unknown')
-            created_time = item.get('createdTimeFormatted', 'Unknown')
-            print(f"  - {item['name']} ({item_type}, Created: {created_time})")
-        if len(all_items) > 3:
-            print(f"  ... and {len(all_items) - 3} more items")
-    else:
-        print("No items found.")
-    print()
-    
-    # Example 5: Check if a file exists
-    print("Example 5: Checking if a file exists")
-    print("----------------------------------")
-    # Use the same file name from Example 3
-    file_to_check = file_name
-    
-    exists = drive.file_exists(file_to_check, subfolder_path)
-    print(f"File '{file_to_check}' {'exists' if exists else 'does not exist'} in '{subfolder_path}'.")
-    print()
-    
-    # Example 6: Get file modified time
-    print("Example 6: Getting file modified time")
-    print("-----------------------------------")
-    # Use the same file name from Example 3
-    file_to_check_time = file_name
-    
-    modified_time = drive.get_file_modified_time(file_to_check_time, subfolder_path)
-    if modified_time:
-        print(f"File '{file_to_check_time}' was last modified on: {modified_time}")
-    else:
-        print(f"Could not get modified time for '{file_to_check_time}'.")
-    print()
-    
-    # Example 7: Get file with content
-    print("Example 7: Getting file with content")
-    print("----------------------------------")
-    # Use the same file name from Example 3
-    file_with_content = file_name
-    
-    file_with_content_obj = drive.get_file(file_with_content, subfolder_path, include_content=True)
-    if file_with_content_obj and 'file_content' in file_with_content_obj:
-        content = file_with_content_obj['file_content']
-        if content:
-            print(f"File '{file_with_content}' content (first 100 chars):")
-            print(f"  {content[:100]}...")
-        else:
-            print(f"File '{file_with_content}' has no content or content could not be read.")
-    else:
-        print(f"File '{file_with_content}' not found or content could not be retrieved.")
-    print()
-    
-    # Example 8: Get contents of all files in a folder
-    print("Example 8: Getting contents of all files in a folder")
-    print("------------------------------------------------")
-    # Get all files with content
-    all_files_with_content = drive.get_files_in_folder(subfolder_path, include_content=True)
-    
-    if all_files_with_content:
-        print(f"Found {len(all_files_with_content)} files. Showing content preview for first 3:")
-        for i, file in enumerate(all_files_with_content[:3]):
-            print(f"  File: {file['name']}")
-            if 'file_content' in file and file['file_content']:
-                content = file['file_content']
-                print(f"    Content preview: {content[:50]}...")
-            else:
-                print(f"    No content available or file is not text-based.")
-        
-        if len(all_files_with_content) > 3:
-            print(f"  ... and {len(all_files_with_content) - 3} more files with content")
-    else:
-        print("No files found or no content could be retrieved.")
-    print()
-    
-    # Example 9: Get content from a specific file using read_file_from_object
-    print("Example 9: Getting content from a specific file using read_file_from_object")
-    print("------------------------------------------------------------------------")
-    # Get a file object first
-    file_obj = drive.get_file(file_name, subfolder_path)
-    
-    if file_obj:
-        # Read the content directly from the file object
-        content = drive.read_file_from_object(file_obj)
-        if content:
-            print(f"File '{file_obj['name']}' content (first 100 chars):")
-            print(f"  {content[:100]}...")
-        else:
-            print(f"File '{file_obj['name']}' has no content or content could not be read.")
-    else:
-        print(f"File '{file_name}' not found.")
-    print()
-    
-    print("=== End of Examples ===")
-
-if __name__ == "__main__":
-    main() 

src/ctp_slack_bot/services/schedule_service.py

@@ -2,6 +2,7 @@ from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
 from asyncio import create_task, iscoroutinefunction, to_thread
 from datetime import datetime
+from dependency_injector.resources import Resource
 from loguru import logger
 from pydantic import BaseModel, PrivateAttr
 from pytz import timezone
@@ -47,11 +48,21 @@ class ScheduleService(BaseModel):
 
     def start(self: Self) -> None:
         self._scheduler.start()
-        logger.info("Started scheduler.")
 
     def stop(self: Self) -> None:
         if self._scheduler.running:
-            self._scheduler.shutdown(wait=False)
-            logger.info("Shut down scheduler.")
+            self._scheduler.shutdown()
         else:
             logger.debug("The scheduler is not running. There is no scheduler to shut down.")
+
+class ScheduleServiceResource(Resource):
+    def init(self: Self, settings: Settings) -> ScheduleService:
+        logger.info("Starting scheduler…")
+        schedule_service = ScheduleService(settings=settings)
+        schedule_service.start()
+        return schedule_service
+
+    def shutdown(self: Self, schedule_service: ScheduleService) -> None:
+        """Stop scheduler on shutdown."""
+        schedule_service.stop()
+        logger.info("Stopped scheduler.")

src/ctp_slack_bot/utils/__init__.py

@@ -0,0 +1 @@
+from ctp_slack_bot.utils.secret_stripper import sanitize_mongo_db_uri

src/ctp_slack_bot/utils/secret_stripper.py

@@ -0,0 +1,6 @@
+from urllib.parse import urlparse, urlunparse
+
+def sanitize_mongo_db_uri(uri: str) -> str:
+    parts = urlparse(uri)
+    sanitized_netloc = ":".join(filter(None, (parts.hostname, parts.port)))
+    return urlunparse((parts.scheme, sanitized_netloc, parts.path, parts.params, parts.query, parts.fragment))