Coverage for src/qdrant_loader/connectors/git/connector.py: 70%
201 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""Git repository connector implementation."""
3import os
4import shutil
5import tempfile
6from collections.abc import AsyncIterator
7from datetime import datetime
9from qdrant_loader.config.types import SourceType
10from qdrant_loader.connectors.base import BaseConnector, resolve_safe_path
11from qdrant_loader.connectors.git.config import GitRepoConfig
12from qdrant_loader.connectors.git.file_processor import FileProcessor
13from qdrant_loader.connectors.git.metadata_extractor import GitMetadataExtractor
14from qdrant_loader.connectors.git.operations import GitOperations
15from qdrant_loader.core.document import Document
16from qdrant_loader.core.file_conversion import (
17 FileConversionConfig,
18 FileConversionError,
19 FileConverter,
20 FileDetector,
21)
22from qdrant_loader.utils.logging import LoggingConfig
24logger = LoggingConfig.get_logger(__name__)
27class GitConnector(BaseConnector):
28 """Git repository connector."""
30 def __init__(self, config: GitRepoConfig):
31 """Initialize the Git connector.
33 Args:
34 config: Configuration for the Git repository
35 """
36 super().__init__(config)
37 self.config = config
38 self.temp_dir = None # Will be set in __enter__
39 self.metadata_extractor = GitMetadataExtractor(config=self.config)
40 self.git_ops = GitOperations()
41 self.file_processor = None # Will be initialized in __enter__
42 self.logger = LoggingConfig.get_logger(__name__)
43 self.logger.debug("Initializing GitConnector")
44 self.logger.debug("GitConnector Configuration", config=config.model_dump())
45 self._initialized = False
47 # Initialize file conversion components if enabled
48 self.file_converter = None
49 self.file_detector = None
50 if self.config.enable_file_conversion:
51 self.logger.debug("File conversion enabled for Git connector")
52 # File conversion config will be set from global config during ingestion
53 self.file_detector = FileDetector()
54 else:
55 self.logger.debug("File conversion disabled for Git connector")
57 def set_file_conversion_config(self, file_conversion_config: FileConversionConfig):
58 """Set file conversion configuration from global config.
60 Args:
61 file_conversion_config: Global file conversion configuration
62 """
63 if self.config.enable_file_conversion:
64 self.file_converter = FileConverter(file_conversion_config)
65 self.logger.debug("File converter initialized with global config")
67 async def __aenter__(self):
68 """Async context manager entry."""
69 try:
70 # Create temporary directory
71 self.temp_dir = tempfile.mkdtemp()
72 self.config.temp_dir = (
73 self.temp_dir
74 ) # Update config with the actual temp dir
75 self.logger.debug("Created temporary directory", temp_dir=self.temp_dir)
77 # Initialize file processor
78 self.file_processor = FileProcessor(
79 config=self.config,
80 temp_dir=self.temp_dir,
81 file_detector=self.file_detector,
82 )
84 # Get auth token from config
85 auth_token = None
86 if self.config.token:
87 auth_token = self.config.token
88 self.logger.debug(
89 "Using authentication token", token_length=len(auth_token)
90 )
92 # Clone repository
93 self.logger.debug(
94 "Attempting to clone repository",
95 url=self.config.base_url,
96 branch=self.config.branch,
97 depth=self.config.depth,
98 temp_dir=self.temp_dir,
99 )
101 try:
102 self.git_ops.clone(
103 url=str(self.config.base_url),
104 to_path=self.temp_dir,
105 branch=self.config.branch,
106 depth=self.config.depth,
107 auth_token=auth_token,
108 )
109 except Exception as clone_error:
110 self.logger.error(
111 "Failed to clone repository",
112 error=str(clone_error),
113 error_type=type(clone_error).__name__,
114 url=self.config.base_url,
115 branch=self.config.branch,
116 temp_dir=self.temp_dir,
117 )
118 raise
120 # Verify repository initialization
121 if not self.git_ops.repo:
122 self.logger.error(
123 "Repository not initialized after clone", temp_dir=self.temp_dir
124 )
125 raise ValueError("Repository not initialized")
127 # Verify repository is valid
128 try:
129 self.git_ops.repo.git.status()
130 self.logger.debug(
131 "Repository is valid and accessible", temp_dir=self.temp_dir
132 )
133 except Exception as status_error:
134 self.logger.error(
135 "Failed to verify repository status",
136 error=str(status_error),
137 error_type=type(status_error).__name__,
138 temp_dir=self.temp_dir,
139 )
140 raise
142 self._initialized = True
143 return self
144 except ValueError as e:
145 # Standardized error logging: user-friendly message + troubleshooting context
146 self.logger.error(
147 "Git repository setup failed due to invalid configuration",
148 error=str(e),
149 error_type="ValueError",
150 suggestion="Verify Git URL format, credentials, and repository accessibility",
151 )
152 raise ValueError(str(e)) from e # Re-raise with the same message
153 except Exception as e:
154 # Standardized error logging: user-friendly message + technical details + cleanup context
155 self.logger.error(
156 "Git repository setup failed during initialization",
157 error=str(e),
158 error_type=type(e).__name__,
159 temp_dir=self.temp_dir,
160 suggestion="Check Git URL, network connectivity, authentication, and disk space",
161 )
162 # Clean up if something goes wrong
163 if self.temp_dir:
164 self._cleanup()
165 raise RuntimeError(f"Failed to set up Git repository: {e}") from e
167 def __enter__(self):
168 """Synchronous context manager entry."""
169 if not self._initialized:
170 self._initialized = True
171 # Create temporary directory
172 self.temp_dir = tempfile.mkdtemp()
173 self.config.temp_dir = (
174 self.temp_dir
175 ) # Update config with the actual temp dir
176 self.logger.debug("Created temporary directory", temp_dir=self.temp_dir)
178 # Initialize file processor
179 self.file_processor = FileProcessor(
180 config=self.config,
181 temp_dir=self.temp_dir,
182 file_detector=self.file_detector,
183 )
185 # Get auth token from config
186 auth_token = None
187 if self.config.token:
188 auth_token = self.config.token
189 self.logger.debug(
190 "Using authentication token", token_length=len(auth_token)
191 )
193 # Clone repository
194 self.logger.debug(
195 "Attempting to clone repository",
196 url=self.config.base_url,
197 branch=self.config.branch,
198 depth=self.config.depth,
199 temp_dir=self.temp_dir,
200 )
202 try:
203 self.git_ops.clone(
204 url=str(self.config.base_url),
205 to_path=self.temp_dir,
206 branch=self.config.branch,
207 depth=self.config.depth,
208 auth_token=auth_token,
209 )
210 except Exception as clone_error:
211 self.logger.error(
212 "Failed to clone repository",
213 error=str(clone_error),
214 error_type=type(clone_error).__name__,
215 url=self.config.base_url,
216 branch=self.config.branch,
217 temp_dir=self.temp_dir,
218 )
219 raise
221 # Verify repository initialization
222 if not self.git_ops.repo:
223 self.logger.error(
224 "Repository not initialized after clone", temp_dir=self.temp_dir
225 )
226 raise ValueError("Repository not initialized")
228 # Verify repository is valid
229 try:
230 self.git_ops.repo.git.status()
231 self.logger.debug(
232 "Repository is valid and accessible", temp_dir=self.temp_dir
233 )
234 except Exception as status_error:
235 self.logger.error(
236 "Failed to verify repository status",
237 error=str(status_error),
238 error_type=type(status_error).__name__,
239 temp_dir=self.temp_dir,
240 )
241 raise
242 return self
244 async def __aexit__(self, exc_type, exc_val, _exc_tb):
245 """Async context manager exit."""
246 self._cleanup()
247 self._initialized = False
249 def __exit__(self, exc_type, exc_val, _exc_tb):
250 """Clean up resources."""
251 self._cleanup()
253 def _cleanup(self):
254 """Clean up temporary directory."""
255 if self.temp_dir and os.path.exists(self.temp_dir):
256 try:
257 shutil.rmtree(self.temp_dir)
258 self.logger.debug("Cleaned up temporary directory")
259 except Exception as e:
260 self.logger.error(f"Failed to clean up temporary directory: {e}")
262 def _process_file(self, file_path: str) -> Document:
263 """Process a single file.
265 Args:
266 file_path: Path to the file
268 Returns:
269 Document instance with file content and metadata
271 Raises:
272 Exception: If file processing fails
273 """
274 try:
275 # Get relative path from repository root
276 rel_path = os.path.relpath(file_path, self.temp_dir)
278 # Fix cross-platform path issues: ensure we get a proper relative path
279 # If relpath returns a path that goes up directories (contains ..),
280 # it means the path calculation failed (common with mixed path styles)
281 if rel_path.startswith("..") and self.temp_dir:
282 # Fallback: try to extract relative path manually
283 if file_path.startswith(self.temp_dir):
284 # Remove temp_dir prefix and any leading separators
285 rel_path = (
286 file_path[len(self.temp_dir) :]
287 .lstrip(os.sep)
288 .lstrip("/")
289 .lstrip("\\")
290 )
291 else:
292 # Last resort: use basename
293 rel_path = os.path.basename(file_path)
295 # Check if file needs conversion
296 needs_conversion = (
297 self.config.enable_file_conversion
298 and self.file_detector
299 and self.file_converter
300 and self.file_detector.is_supported_for_conversion(file_path)
301 )
303 if needs_conversion:
304 self.logger.debug("File needs conversion", file_path=rel_path)
305 try:
306 # Convert file to markdown
307 assert self.file_converter is not None # Type checker hint
308 content = self.file_converter.convert_file(file_path)
309 content_type = "md" # Converted files are markdown
310 conversion_method = "markitdown"
311 conversion_failed = False
312 self.logger.info("File conversion successful", file_path=rel_path)
313 except FileConversionError as e:
314 self.logger.warning(
315 "File conversion failed, creating fallback document",
316 file_path=rel_path,
317 error=str(e),
318 )
319 # Create fallback document
320 assert self.file_converter is not None # Type checker hint
321 content = self.file_converter.create_fallback_document(file_path, e)
322 content_type = "md" # Fallback is also markdown
323 conversion_method = "markitdown_fallback"
324 conversion_failed = True
325 else:
326 # Read file content normally
327 content = self.git_ops.get_file_content(file_path)
328 # Get file extension without the dot
329 content_type = os.path.splitext(file_path)[1].lower().lstrip(".")
330 conversion_method = None
331 conversion_failed = False
333 first_commit_date = self.git_ops.get_first_commit_date(file_path)
335 # Get last commit date
336 last_commit_date = self.git_ops.get_last_commit_date(file_path)
338 # Extract metadata
339 metadata = self.metadata_extractor.extract_all_metadata(
340 file_path=rel_path, content=content
341 )
343 # Add Git-specific metadata
344 metadata.update(
345 {
346 "repository_url": self.config.base_url,
347 "branch": self.config.branch,
348 "last_commit_date": (
349 last_commit_date.isoformat() if last_commit_date else None
350 ),
351 }
352 )
354 # Add file conversion metadata if applicable
355 if needs_conversion:
356 metadata.update(
357 {
358 "conversion_method": conversion_method,
359 "conversion_failed": conversion_failed,
360 "original_file_type": os.path.splitext(file_path)[1]
361 .lower()
362 .lstrip("."),
363 }
364 )
366 self.logger.debug(f"Processed Git file: /{rel_path!s}")
368 # Create document
369 # Normalize path separators for URL (use forward slashes on all platforms)
370 normalized_rel_path = rel_path.replace(os.sep, "/").replace("\\", "/")
371 git_document = Document(
372 title=os.path.basename(file_path),
373 content=content,
374 content_type=content_type,
375 metadata=metadata,
376 source_type=SourceType.GIT,
377 source=self.config.source,
378 url=f"{str(self.config.base_url).replace('.git', '')}/blob/{self.config.branch}/{normalized_rel_path}",
379 is_deleted=False,
380 created_at=first_commit_date,
381 updated_at=last_commit_date,
382 )
384 return git_document
385 except Exception as e:
386 self.logger.error(
387 "Failed to process file", file_path=file_path, error=str(e)
388 )
389 raise
391 async def stream_documents(
392 self, since: datetime | None = None
393 ) -> AsyncIterator[Document]:
394 """Stream documents from the repository (WS-1 connector contract).
396 Note:
397 The `since` parameter is not yet implemented for incremental
398 ingestion. All files in the repository are processed regardless
399 of modification time.
401 Yields:
402 Document objects from the repository.
404 Raises:
405 Exception: If document retrieval fails
406 """
407 try:
408 self._ensure_initialized()
409 try:
410 files = (
411 self.git_ops.list_files()
412 ) # This will raise ValueError if not initialized
413 except ValueError as e:
414 self.logger.error("Failed to list files", error=str(e))
415 raise ValueError("Repository not initialized") from e
417 for file_path in files:
418 if not self.file_processor.should_process_file(file_path): # type: ignore
419 continue
421 try:
422 document = self._process_file(file_path)
423 yield document
425 except Exception as e:
426 self.logger.error(
427 "Failed to process file", file_path=file_path, error=str(e)
428 )
429 continue
431 except ValueError as e:
432 # Re-raise ValueError to maintain the error type
433 self.logger.error("Failed to get documents", error=str(e))
434 raise
435 except Exception as e:
436 self.logger.error("Failed to get documents", error=str(e))
437 raise
439 async def get_documents(self) -> list[Document]:
440 """Get documents from the repository (DEPRECATED - use stream_documents)."""
441 return await super().get_documents()
443 async def fetch_by_id(self, entity_id: str) -> Document | None:
444 """Fetch a single file by its repo-relative path (forward slashes)."""
445 self._ensure_initialized()
446 file_path = resolve_safe_path(self.temp_dir, entity_id)
447 if file_path is None:
448 self.logger.warning("Path traversal attempt blocked", entity_id=entity_id)
449 return None
450 if not os.path.exists(file_path) or not self.file_processor.should_process_file( # type: ignore
451 file_path
452 ):
453 return None
454 try:
455 return self._process_file(file_path)
456 except Exception as e:
457 self.logger.error(
458 "Failed to fetch file by id", entity_id=entity_id, error=str(e)
459 )
460 return None
462 async def list_entity_ids(self) -> AsyncIterator[str]:
463 """Stream repo-relative paths (forward slashes) for all processable files."""
464 self._ensure_initialized()
465 for file_path in self.git_ops.list_files():
466 if not self.file_processor.should_process_file(file_path): # type: ignore
467 continue
468 rel_path = os.path.relpath(file_path, self.temp_dir)
469 yield rel_path.replace(os.sep, "/").replace("\\", "/")
471 def _ensure_initialized(self):
472 """Ensure the repository is initialized before performing operations."""
473 if not self._initialized:
474 self.logger.error(
475 "Repository not initialized. Use the connector as a context manager."
476 )
477 raise ValueError("Repository not initialized")