Introduce dependency injection

src/ctp_slack_bot/api/main.py

@@ -1,14 +1,15 @@
 from contextlib import asynccontextmanager
-from fastapi import FastAPI, HTTPException
+from fastapi import FastAPI, HTTPException, Depends
 from loguru import logger
-from typing import AsyncGenerator, Never
+from typing import AsyncGenerator
+from dependency_injector.wiring import inject, Provide
 
 from ctp_slack_bot.api.routes import router
-from ctp_slack_bot.core.config import Settings, settings
+from ctp_slack_bot.core.config import Settings
 from ctp_slack_bot.core.logging import setup_logging
 from ctp_slack_bot.core.response_rendering import PrettyJSONResponse
 from ctp_slack_bot.tasks.scheduler import start_scheduler, stop_scheduler
-
+from ctp_slack_bot.containers import Container
 
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator:
@@ -16,20 +17,31 @@ async def lifespan(app: FastAPI) -> AsyncGenerator:
     Lifespan context manager for FastAPI application.
     Handles startup and shutdown events.
     """
-    # Setup logging
-    setup_logging()
+    # Initialize container and wire the container to modules that need dependency injection.
+    container = Container()
+    app.container = container
+    container.wire(
+        modules=[
+            "ctp_slack_bot.api.routes",
+            "ctp_slack_bot.services",
+            "ctp_slack_bot.tasks"
+        ]
+    )
+
+    # Setup logging.
+    setup_logging(container)
     logger.info("Starting application")
-    
-    # Start scheduler
-    #scheduler = start_scheduler()
-    #logger.info("Started scheduler")
-    
-    yield
-    
-    # Shutdown
+
+    # Start the scheduler.
+    scheduler = start_scheduler(container)
+    logger.info("Started scheduler")
+
+    yield # control to FastAPI until shutdown.
+
+    # Shutdown.
     logger.info("Shutting down application")
-    #stop_scheduler(scheduler)
-    #logger.info("Stopped scheduler")
+    stop_scheduler(scheduler)
+    logger.info("Stopped scheduler")
 
 
 app = FastAPI(
@@ -39,30 +51,27 @@ app = FastAPI(
     lifespan=lifespan,
 )
 
-# Include routers
+# Include routers.
 app.include_router(router)
 
+# Provide a minimalist health check endpoint for clients to detect availability.
 @app.get("/health")
-async def health() -> dict[str, str]:
+async def get_health() -> dict[str, str]:
     """Health check"""
     return {
         "status": "healthy"
     }
 
-@app.get("/env", response_class=PrettyJSONResponse)
-async def env() -> Settings:
-    """Server-internal environment variables"""
-    if not settings.DEBUG:
-        raise HTTPException(status_code=404)
-    return settings
-
-
-if __name__ == "__main__":
+# Alternate starting path for development
+def run() -> None:
     import uvicorn
-    
+    settings = Settings() # type: ignore
     uvicorn.run(
         "main:app",
         host=settings.API_HOST,
         port=settings.API_PORT,
         reload=settings.DEBUG
     )
+
+if __name__ == "__main__":
+    run()

src/ctp_slack_bot/api/routes.py

@@ -1,13 +1,20 @@
 from fastapi import APIRouter, Depends, HTTPException, status
+from dependency_injector.wiring import inject, Provide
 from loguru import logger
 
+from ctp_slack_bot.containers import Container
 #from ctp_slack_bot.api.dependencies import get_slack_service, get_transcript_service
+from ctp_slack_bot.core.config import Settings
 #from ctp_slack_bot.models.transcript import TranscriptRequest, TranscriptResponse
 #from ctp_slack_bot.services.slack_service import SlackService
 #from ctp_slack_bot.services.transcript_service import TranscriptService
 
 router = APIRouter(prefix="/api/v1")
 
+@router.get("/env", response_model=Settings)
+@inject
+async def get_env(settings: Settings = Depends(Provide[Container.settings])) -> Settings:
+    return settings
 
 # @router.post("/transcripts/analyze", response_model=TranscriptResponse)
 # async def analyze_transcript(

src/ctp_slack_bot/containers.py

@@ -0,0 +1,59 @@
+from dependency_injector.containers import DeclarativeContainer
+from dependency_injector.providers import Configuration, Factory, Singleton
+#from pymongo import MongoClient
+#from langchain.embeddings import HuggingFaceEmbeddings
+#from transformers import AutoTokenizer, AutoModel
+
+from ctp_slack_bot.core.config import Settings
+#from ctp_slack_bot.db.connection import get_database
+#from ctp_slack_bot.services.vectorization_service import VectorizationService
+
+class Container(DeclarativeContainer):
+    # Core dependencies
+    settings = Singleton(Settings)
+
+    # db_client = Singleton(
+    #     MongoClient,
+    #     host=config.MONGODB_URI.get_secret_value
+    # )
+
+    # db = Singleton(
+    #     get_database,
+    #     client=db_client,
+    #     db_name=config.MONGODB_NAME
+    # )
+
+    # Machine-learning models
+    # tokenizer = Singleton(
+    #     AutoTokenizer.from_pretrained,
+    #     config.EMBEDDING_MODEL
+    # )
+
+    # model = Singleton(
+    #     AutoModel.from_pretrained,
+    #     config.EMBEDDING_MODEL
+    # )
+
+    # embeddings = Singleton(
+    #     HuggingFaceEmbeddings,
+    #     model_name=config.EMBEDDING_MODEL
+    # )
+
+    # Repositories
+    # transcript_repository = Factory(
+    #     # Your transcript repository class
+    #     db=db
+    # )
+
+    # Services
+    # vectorization_service = Factory(
+    #     VectorizationService,
+    #     embeddings=embeddings,
+    #     chunk_size=config.CHUNK_SIZE,
+    #     chunk_overlap=config.CHUNK_OVERLAP,
+    #     vector_dimension=config.VECTOR_DIMENSION
+    # )
+
+    # Add other services here
+    # transcript_service = providers.Factory(...)
+    # slack_service = providers.Factory(...)

src/ctp_slack_bot/core/config.py

@@ -1,9 +1,6 @@
-from functools import lru_cache
-from typing import Literal, Optional
-
 from pydantic import Field, MongoDsn, NonNegativeFloat, NonNegativeInt, PositiveInt, SecretStr
 from pydantic_settings import BaseSettings, SettingsConfigDict
-
+from typing import Literal, Optional
 
 class Settings(BaseSettings): # TODO: Strong guarantees of validity, because garbage in = garbage out, and settings flow into all the nooks and crannies
     """
@@ -54,14 +51,3 @@ class Settings(BaseSettings): # TODO: Strong guarantees of validity, because gar
         env_file_encoding="utf-8",
         case_sensitive=True,
     )
-
-
-@lru_cache
-def get_settings() -> Settings:
-    """
-    Get cached settings instance.
-    """
-    return Settings() # type: ignore
-
-
-settings = get_settings()

src/ctp_slack_bot/core/logging.py

@@ -1,13 +1,12 @@
-import logging
-import sys
-from typing import Dict, Union
-
+from logging import __file__ as logging_file, basicConfig, currentframe, getLogger, Handler, INFO, LogRecord
 from loguru import logger
+from sys import stderr
+from typing import Dict, Union
 
-from ctp_slack_bot.core.config import settings
+from ctp_slack_bot.containers import Container
 
 
-class InterceptHandler(logging.Handler):
+class InterceptHandler(Handler):
     """
     Intercept standard logging messages toward Loguru.
     
@@ -15,7 +14,7 @@ class InterceptHandler(logging.Handler):
     to Loguru, allowing unified logging across the application.
     """
     
-    def emit(self, record: logging.LogRecord) -> None:
+    def emit(self, record: LogRecord) -> None:
         # Get corresponding Loguru level if it exists
         try:
             level = logger.level(record.levelname).name
@@ -23,8 +22,8 @@ class InterceptHandler(logging.Handler):
             level = record.levelno
         
         # Find caller from where the logged message originated
-        frame, depth = logging.currentframe(), 2
-        while frame and frame.f_code.co_filename == logging.__file__:
+        frame, depth = currentframe(), 2
+        while frame and frame.f_code.co_filename == logging_file:
             frame = frame.f_back
             depth += 1
         
@@ -33,7 +32,7 @@ class InterceptHandler(logging.Handler):
         )
 
 
-def setup_logging() -> None:
+def setup_logging(container: Container) -> None:
     """
     Configure logging with Loguru.
     
@@ -41,9 +40,11 @@ def setup_logging() -> None:
     configures the log format based on settings, and intercepts
     standard logging messages.
     """
+    settings = container.settings() if container else Provide[Container.settings]
+
     # Remove default loguru handler
     logger.remove()
-    
+
     # Determine log format
     if settings.LOG_FORMAT == "json":
         log_format = {
@@ -62,17 +63,17 @@ def setup_logging() -> None:
             "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
             "<level>{message}</level>"
         )
-    
+
     # Add console handler
     logger.add(
-        sys.stderr,
+        stderr,
         format=format_string,
         level=settings.LOG_LEVEL,
         serialize=(settings.LOG_FORMAT == "json"),
         backtrace=True,
         diagnose=True,
     )
-    
+
     # Add file handler for non-DEBUG environments
     if settings.LOG_LEVEL != "DEBUG":
         logger.add(
@@ -84,19 +85,12 @@ def setup_logging() -> None:
             level=settings.LOG_LEVEL,
             serialize=(settings.LOG_FORMAT == "json"),
         )
-    
+
     # Intercept standard logging messages
-    logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
-    
+    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",
-    ]:
-        logging.getLogger(logger_name).setLevel(logging.INFO)
-    
+    for logger_name in ("uvicorn", "uvicorn.error", "fastapi", "httpx", "apscheduler", "pymongo"):
+        getLogger(logger_name).setLevel(INFO)
+
     logger.info(f"Logging configured with level {settings.LOG_LEVEL}")

src/ctp_slack_bot/tasks/scheduler.py

@@ -1,28 +1,28 @@
-from datetime import datetime
-from typing import Optional
-
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
+from datetime import datetime
+from dependency_injector.wiring import inject, Provide
 from loguru import logger
 from pytz import timezone
+from typing import Optional
 
-from ctp_slack_bot.core.config import settings
+from ctp_slack_bot.containers import Container
 #from ctp_slack_bot.tasks.error_report import send_error_report
 #from ctp_slack_bot.tasks.transcript_cleanup import cleanup_old_transcripts
 
-
-def start_scheduler() -> AsyncIOScheduler:
+@inject
+def start_scheduler(container: Container) -> AsyncIOScheduler:
     """
     Start and configure the APScheduler instance.
     
     Returns:
         AsyncIOScheduler: Configured scheduler instance
     """
-    scheduler = AsyncIOScheduler(timezone=timezone(settings.SCHEDULER_TIMEZONE))
+    settings = container.settings() if container else Provide[Container.settings]
+    zone = settings.SCHEDULER_TIMEZONE
+    scheduler = AsyncIOScheduler(timezone=timezone(zone))
     
-    # Add jobs to the scheduler
-    
-    # Daily error report at 7 AM
+    # Add jobs to the scheduler.
     # scheduler.add_job(
     #     send_error_report,
     #     CronTrigger(hour=7, minute=0),
@@ -30,8 +30,6 @@ def start_scheduler() -> AsyncIOScheduler:
     #     name="Daily Error Report",
     #     replace_existing=True,
     # )
-    
-    # Weekly transcript cleanup on Sundays at 1 AM
     # scheduler.add_job(
     #     cleanup_old_transcripts,
     #     CronTrigger(day_of_week="sun", hour=1, minute=0),
@@ -40,25 +38,25 @@ def start_scheduler() -> AsyncIOScheduler:
     #     replace_existing=True,
     # )
     
-    # Start the scheduler
+    # Start the scheduler.
     scheduler.start()
     logger.info("Scheduler started with timezone: {}", settings.SCHEDULER_TIMEZONE)
-    logger.info("Next run for error report: {}", 
-                scheduler.get_job("daily_error_report").next_run_time)
-    logger.info("Next run for transcript cleanup: {}", 
-                scheduler.get_job("weekly_transcript_cleanup").next_run_time)
+    # logger.info("Next run for error report: {}", 
+    #             scheduler.get_job("daily_error_report").next_run_time)
+    # logger.info("Next run for transcript cleanup: {}", 
+    #             scheduler.get_job("weekly_transcript_cleanup").next_run_time)
     
     return scheduler
 
 
-def stop_scheduler(scheduler: Optional[AsyncIOScheduler] = None) -> None:
+def stop_scheduler(scheduler: AsyncIOScheduler) -> None:
     """
     Shutdown the scheduler gracefully.
     
     Args:
         scheduler: The scheduler instance to shut down
     """
-    if scheduler is not None and scheduler.running:
+    if scheduler.running:
         logger.info("Shutting down scheduler")
         scheduler.shutdown(wait=False)
         logger.info("Scheduler shutdown complete")