embedder_wrapper.py

import asyncio
import sys
import os
from typing import List

# Add the src/python directory to the path so we can import our modules
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src', 'python'))

from src.python.embedder import Embedder


class SyncEmbedder:
    """
    A synchronous wrapper for the async Embedder class.
    """
    
    def __init__(self):
        self.async_embedder = Embedder()
    
    def embed_text(self, text: str) -> List[float]:
        """
        Synchronous method to create an embedding for a single text.
        """
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            embedding = loop.run_until_complete(self.async_embedder.create_embedding(text))
            return embedding
        finally:
            loop.close()
    
    def embed_texts(self, texts: List[str]) -> List[List[float]]:
        """
        Synchronous method to create embeddings for a list of texts.
        """
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        try:
            embeddings = loop.run_until_complete(self.async_embedder.create_embeddings_batch(texts))
            return embeddings
        finally:
            loop.close()