Newer
Older
cortex-hub / ai-hub / app / api / dependencies.py
# app/api/dependencies.py
from fastapi import Depends, HTTPException, status
from typing import List,Any
from sqlalchemy.orm import Session
from app.db.session import SessionLocal
from app.core.retrievers.base_retriever import Retriever
from app.core.services.document import DocumentService
from app.core.services.rag import RAGService
from app.core.vector_store.faiss_store import FaissVectorStore


# This is a dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# This is another common dependency
async def get_current_user(token: str):
    # In a real app, you would decode the token and fetch the user
    if not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return {"email": "user@example.com", "id": 1} # Dummy user


class ServiceContainer:
    """
    A flexible container for managing and providing various application services.
    Services are added dynamically using the `with_service` method.
    """
    def __init__(self):
        # Use a dictionary to store services, mapping their names to instances
        self._services = {}
        self.document_service = None
        self.rag_service = None

    def with_service(self, name: str, service: Any):
        """
        Adds a service to the container.

        Args:
            name (str): The name to assign to the service (e.g., 'tts_service').
            service (Any): The service instance to add.
        """
        setattr(self, name, service)
        return self

    def with_document_service(self, vector_store: FaissVectorStore):
        """
        Adds a DocumentService instance to the container.
        """
        self.document_service = DocumentService(vector_store=vector_store)
        return self
    
    def with_rag_service(self, retrievers: List[Retriever]):
        """
        Adds a RAGService instance to the container.
        """
        self.rag_service = RAGService(retrievers=retrievers)
        return self

    def __getattr__(self, name: str) -> Any:
        """
        Allows services to be accessed directly as attributes (e.g., container.rag_service).
        """
        # This allows direct access to services that are not explicitly defined in __init__
        try:
            return self.__getattribute__(name)
        except AttributeError:
            raise AttributeError(f"'{self.__class__.__name__}' object has no service named '{name}'")