Dependency Injection Pattern

File: app/core/dependencies.py Status: ✅ Implemented and Working Purpose: Centralized service registry with singleton pattern

---

Overview

The dependency injection pattern ensures: 1. Single instances of services (singletons) created at startup 2. Proper dependency wiring (e.g., RetrievalPipeline depends on GPTMiniService) 3. FastAPI integration via Depends() pattern 4. Testability (can mock dependencies in tests)

---

Service Initialization Order

Services are initialized in dependency order:

# 1. External Clients (no dependencies)
openai_client = OpenAI(api_key=settings.OPENAI_API_KEY)
supabase_service = create_client(settings.SUPABASE_URL, settings.SUPABASE_SERVICE_ROLE_KEY)

# 2. Base Adapters (depend on clients)
pinecone_adapter = PineconeAdapter(api_key=..., index_name=...)
cache_service = CacheService()

# 3. Core Services (depend on adapters)
embedding_service = EmbeddingService(
    openai_client=openai_client,
    supabase=supabase_service,
    pinecone_adapter=pinecone_adapter
)

gpt_mini_service = GPTMiniService(
    openai_client=openai_client,
    model=settings.OPENAI_MINI_MODEL
)

# 4. Pipelines (depend on core services)
retrieval_pipeline = RetrievalPipeline(
    openai_client=openai_client,
    supabase=supabase_service,
    pinecone_adapter=pinecone_adapter,
    embedding_service=embedding_service,
    gpt_mini_service=gpt_mini_service,
    cache_service=cache_service
)

# 5. Feature Services (depend on pipelines)
quiz_generator = QuizGeneratorService(
    openai_client=openai_client,
    retrieval_pipeline=retrieval_pipeline
)

---

Dependency Graph

graph TD
    OAI[OpenAI Client] --> ES[EmbeddingService]
    OAI --> GPT[GPTMiniService]
    OAI --> QG[QuizGeneratorService]

    SB[Supabase Client] --> ES
    SB --> WS[WalletReservationService]
    SB --> IS[IngestionService]
    SB --> SS[ScraperService]
    SB --> RP[RetrievalPipeline]

    PC[Pinecone Adapter] --> ES
    PC --> RP

    CS[CacheService] --> RP

    ES --> RP
    GPT --> RP

    RP --> QG

    TN[TextNormalizer] --> SS
    DS[DeduplicationService] --> SS
    QC[QualityChecker] --> SS

---

Usage in Routers

Pattern

# 1. Import singleton from dependencies
from app.core.dependencies import wallet_service, retrieval_pipeline

# 2. Use in endpoint with FastAPI Depends
@router.post("/ask")
async def ask_question(
    request: AskRequest,
    user: dict = Depends(get_current_user)
):
    # 3. Call service methods
    balance = wallet_service.get_balance(user["id"])

    # 4. Use service
    if balance["token_balance"] < estimated_cost:
        raise HTTPException(402, "Insufficient balance")

    reservation = wallet_service.reserve(user["id"], estimated_cost)

    # ... rest of logic

Working Example (Wallet Router)

File: app/api/routers/wallet.py

from app.core.dependencies import wallet_service, supabase_service

@router.get("/balance", response_model=WalletBalanceResponse)
async def get_balance(user: dict = Depends(get_current_user)):
    """Get wallet balance."""
    # Uses singleton wallet_service
    res = supabase_service.table("wallet").select("*").eq("user_id", user["id"]).single().execute()

    # Count pending reservations
    pending_res = supabase_service.table("reservations").select("*", count="exact").eq("user_id", user["id"]).eq("status", "reserved").execute()

    return WalletBalanceResponse(
        user_id=UUID(user["id"]),
        token_balance=res.data["token_balance"],
        subscription_tier=res.data["subscription_tier"],
        pending_reservations=pending_res.count or 0
    )

TODO: Chat Router Integration

File: app/api/routers/chat.py (currently stub)

Needs:

from app.core.dependencies import (
    retrieval_pipeline,
    wallet_service,
    gpt_mini_service
)

@router.post("/ask", response_model=AskResponse)
async def ask_question(
    request: AskRequest,
    user: dict = Depends(get_current_user)
):
    # 1. Validate input
    validation = gpt_mini_service.validate_input(request.question)
    if not validation["safe"]:
        raise HTTPException(400, "Unsafe content")

    # 2. Estimate cost
    from app.services.tier_config import TierConfig
    balance = wallet_service.get_balance(UUID(user["id"]))
    tier = balance["subscription_tier"]
    estimated = TierConfig.calculate_estimated_cost(tier, len(request.question))

    # 3. Reserve tokens
    reservation = wallet_service.reserve(
        user_id=UUID(user["id"]),
        estimated=estimated,
        request_id=UUID(get_request_id())
    )

    # 4. Execute retrieval
    retrieval_result = retrieval_pipeline.retrieve(
        query=request.question,
        user_tier=tier,
        namespace=f"grade-{request.grade}-{request.subject}",
        filter={"language": request.language or "fr"}
    )

    # 5. Generate answer with GPT-4o
    # ... (use retrieval_result["results"] as context)

    # 6. Finalize reservation
    wallet_service.finalize(
        reservation_id=UUID(reservation["id"]),
        actual=actual_tokens_used
    )

    # 7. Return response
    return AskResponse(...)

---

Service Registry

All available singletons in app.core.dependencies:

Service	Variable Name	Used By
OpenAI Client	openai_client	embedding_service, gpt_mini_service, quiz_generator
Supabase Service	supabase_service	All services needing database access
Pinecone Adapter	pinecone_adapter	embedding_service, retrieval_pipeline
Cache Service	cache_service	retrieval_pipeline
Embedding Service	embedding_service	retrieval_pipeline, ingestion workers
GPT-mini Service	gpt_mini_service	retrieval_pipeline
Retrieval Pipeline	retrieval_pipeline	chat router, quiz_generator
Wallet Service	wallet_service	wallet router, chat router, quiz router
Chunking Service	chunking_service	ingestion workers
Ingestion Service	ingestion_service	admin router, ingestion router
Text Normalizer	text_normalizer	scraper_service
Deduplication Service	deduplication_service	scraper_service
Quality Checker	quality_checker	scraper_service
Scraper Service	scraper_service	scraper router
Quiz Generator	quiz_generator	quiz router

---

Testing with Dependency Injection

Unit Tests

Mock the dependencies:

from unittest.mock import Mock
from app.services.wallet_reservation import WalletReservationService

def test_reserve_insufficient_balance():
    # Mock Supabase client
    mock_supabase = Mock()
    mock_supabase.table().select().eq().execute().data = [
        {"token_balance": 5}  # Insufficient
    ]

    # Create service with mock
    service = WalletReservationService(supabase=mock_supabase)

    # Test
    with pytest.raises(InsufficientBalanceError):
        service.reserve(user_id=UUID(...), estimated=10)

Integration Tests

Override dependencies:

from fastapi.testclient import TestClient
from app.main import app
from app.core.dependencies import wallet_service

# Create mock service
mock_wallet = Mock()
mock_wallet.get_balance.return_value = {"token_balance": 100, ...}

# Override dependency
app.dependency_overrides[wallet_service] = lambda: mock_wallet

# Test
client = TestClient(app)
response = client.get("/wallet/balance", headers={"Authorization": "Bearer ..."})
assert response.status_code == 200

---

Benefits of This Pattern

1. Singleton Pattern

• Services initialized once at startup
• Shared state (cache, circuit breaker) across all requests
• No repeated initialization overhead

2. Dependency Inversion

• Routers depend on abstractions (service interfaces)
• Easy to swap implementations (e.g., in-memory cache → Redis)
• Testable (inject mocks)

3. Clear Dependencies

• Explicit dependency graph
• Easy to understand service relationships
• Prevents circular dependencies

4. FastAPI Integration

• Works with Depends() for automatic injection
• Type hints provide autocomplete
• Validation at compile time

---

Configuration

Services read from app.core.config.Settings:

# From .env
settings.OPENAI_API_KEY          → openai_client
settings.PINECONE_API_KEY        → pinecone_adapter
settings.SUPABASE_URL            → supabase_service
settings.SUPABASE_SERVICE_ROLE_KEY → supabase_service
settings.OPENAI_EMBEDDING_MODEL  → embedding_service
settings.OPENAI_MINI_MODEL       → gpt_mini_service

---

Future Improvements

1. Lifespan Management

Use FastAPI lifespan events:

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Initialize services
    from app.core import dependencies
    yield
    # Shutdown: Close connections
    # (Close OpenAI client, Pinecone, etc.)

app = FastAPI(lifespan=lifespan)

2. Health Checks for Dependencies

@router.get("/health/dependencies")
async def check_dependencies():
    return {
        "openai": check_openai_health(),
        "supabase": check_supabase_health(),
        "pinecone": check_pinecone_health()
    }

3. Dependency Injection Framework

Consider using dependency-injector library:

from dependency_injector import containers, providers

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    openai_client = providers.Singleton(
        OpenAI,
        api_key=config.openai_api_key
    )

    wallet_service = providers.Singleton(
        WalletReservationService,
        supabase=supabase_service
    )

---

Status: ✅ Dependency injection working for wallet, admin, auth routers TODO: Wire chat, quiz, ingestion, scraper routers

---

See implementation_status.md for current implementation state