Coverage for src / qdrant_loader / connectors / base.py: 100%

1from abc import ABC, abstractmethod 

2 

3from qdrant_loader.config.source_config import SourceConfig 

4from qdrant_loader.core.document import Document 

5from qdrant_loader.core.file_conversion import FileConversionConfig 

6 

7 

8class ConnectorConfigurationError(Exception): 

9 """Raised when a connector's configuration is invalid or access is denied. 

10 

11 This is a *fatal* error: the pipeline should stop rather than silently 

12 continuing with 0 documents. 

13 """ 

14 

15 

16class BaseConnector(ABC): 

17 """Base class for all connectors.""" 

18 

19 def __init__(self, config: SourceConfig): 

20 self.config = config 

21 self._initialized = False 

22 

23 async def __aenter__(self): 

24 """Async context manager entry.""" 

25 self._initialized = True 

26 return self 

27 

28 async def __aexit__(self, exc_type, exc_val, _exc_tb): 

29 """Async context manager exit.""" 

30 self._initialized = False 

31 

32 def set_file_conversion_config( 

33 self, file_conversion_config: FileConversionConfig 

34 ) -> None: 

35 """Set file conversion configuration. 

36 

37 This default implementation stores the configuration for potential 

38 use by subclasses that choose to honor it. 

39 

40 Args: 

41 file_conversion_config: Global file conversion configuration 

42 """ 

43 # Store on the instance so connectors that opt-in can access it. 

44 self._file_conversion_config = file_conversion_config 

45 

46 @abstractmethod 

47 async def get_documents(self) -> list[Document]: 

48 """Get documents from the source."""