Coverage for src/qdrant_loader_core/graph/extractor/git.py: 100%

1from __future__ import annotations 

2 

3from typing import TYPE_CHECKING 

4 

5if TYPE_CHECKING: 

6 from qdrant_loader.core.document import Document 

7 

8from qdrant_loader_core.graph.extractor.base_extractor import ( 

9 BaseEntityExtractor, 

10) 

11from qdrant_loader_core.graph.models import ( 

12 CoreNodeLabel, 

13 GraphNode, 

14 PersonInfo, 

15) 

16 

17 

18class GitEntityExtractor(BaseEntityExtractor): 

19 """ 

20 Git repository graph extractor. 

21 

22 Extracts: 

23 - Document node (file) 

24 - Person node from last_commit_author 

25 - Container node from repository 

26 """ 

27 

28 source_type = "git" 

29 

30 @staticmethod 

31 def _repository_id(doc: Document) -> str | None: 

32 name = doc.metadata.get("repository_name") 

33 if not name: 

34 return None 

35 owner = doc.metadata.get("repository_owner") 

36 return f"{owner}/{name}" if owner else name 

37 

38 def _project( 

39 self, 

40 doc: Document, 

41 ) -> str | None: 

42 return self._repository_id(doc) 

43 

44 def _build_document_node( 

45 self, 

46 doc: Document, 

47 project: str | None, 

48 ) -> GraphNode: 

49 return GraphNode( 

50 id=doc.id, 

51 label=CoreNodeLabel.DOCUMENT.value, 

52 project=project, 

53 properties={ 

54 "title": doc.title, 

55 "source_type": self.source_type, 

56 }, 

57 ) 

58 

59 def _extract_people( 

60 self, 

61 doc: Document, 

62 ) -> list[tuple[PersonInfo, str]]: 

63 author = doc.metadata.get("last_commit_author") 

64 

65 if not author: 

66 return [] 

67 

68 return [ 

69 ( 

70 PersonInfo( 

71 id=f"git_user:{author}", 

72 display_name=author, 

73 ), 

74 "last_committer", 

75 ) 

76 ] 

77 

78 def _extract_container( 

79 self, 

80 doc: Document, 

81 ) -> GraphNode | None: 

82 repo_name = self._repository_id(doc) 

83 

84 if not repo_name: 

85 return None 

86 

87 return GraphNode( 

88 id=f"git:{repo_name}", 

89 label=CoreNodeLabel.CONTAINER.value, 

90 project=repo_name, 

91 properties={ 

92 "kind": "repository", 

93 "repository_name": repo_name, 

94 "repository_owner": doc.metadata.get("repository_owner"), 

95 "repository_url": doc.metadata.get("repository_url"), 

96 }, 

97 )