Coverage for src / lexigram / contracts / ai / loaders.py: 0%

1"""Document loader protocols and exceptions.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

6 

7if TYPE_CHECKING: 

8 from lexigram.contracts.ai.vector import Document 

9 from lexigram.contracts.core.result import Result 

10 

11from lexigram.contracts.ai.exceptions import RAGError 

12 

13 

14class LoaderError(RAGError): 

15 """Raised when document loading fails in an expected, recoverable way. 

16 

17 Extended in lexigram-ai-rag with specific failures like unsupported 

18 format, parse errors, network failures, etc. 

19 """ 

20 

21 _code = "LEX_ERR_LOAD_001" 

22 

23 def __init__(self, message: str = "Loader error", **kwargs: Any) -> None: 

24 super().__init__(message, **kwargs) 

25 

26 

27@runtime_checkable 

28class DocumentLoaderProtocol(Protocol): 

29 """Protocol for document loaders. 

30 

31 Implementations load documents from various sources (files, URLs, 

32 databases) and return them as Document objects. 

33 """ 

34 

35 async def load( 

36 self, source: str, **kwargs: Any 

37 ) -> Result[list[Document], LoaderError]: 

38 """Load documents from a source. 

39 

40 Args: 

41 source: File path, URL, or other source identifier. 

42 **kwargs: Loader-specific parameters. 

43 

44 Returns: 

45 Ok(list of Document) on success. 

46 Err(LoaderError) on failure. 

47 """ 

48 ... 

49 

50 

51@runtime_checkable 

52class LoaderRegistryProtocol(Protocol): 

53 """Protocol for a registry that maps sources to loaders. 

54 

55 Implementations maintain a mapping of source types (file extensions, 

56 URL schemes) to loader instances. Used for auto-detection and 

57 on-demand loader resolution. 

58 """ 

59 

60 def register(self, schemes: list[str], loader: DocumentLoaderProtocol) -> None: 

61 """Register a loader for one or more source schemes. 

62 

63 Args: 

64 schemes: List of file extensions (e.g. [".pdf"]) or URL schemes. 

65 loader: DocumentLoaderProtocol instance to handle those schemes. 

66 """ 

67 ... 

68 

69 def get(self, source: str) -> DocumentLoaderProtocol | None: 

70 """Get the loader for a source, or None if not registered. 

71 

72 Args: 

73 source: File path or URL. 

74 

75 Returns: 

76 The registered loader, or None if no match found. 

77 """ 

78 ... 

79 

80 def list(self) -> list[str]: 

81 """List all registered source schemes. 

82 

83 Returns: 

84 List of registered scheme strings. 

85 """ 

86 ... 

87 

88 

89__all__ = [ 

90 "DocumentLoaderProtocol", 

91 "LoaderError", 

92 "LoaderRegistryProtocol", 

93]