Coverage for src/qdrant_loader/cli/commands/ingest.py: 21%

1from __future__ import annotations 

2 

3import traceback 

4from typing import Any 

5 

6from qdrant_loader.utils.logging import LoggingConfig 

7from qdrant_loader.utils.sensitive import sanitize_exception_message 

8 

9 

10async def run_pipeline_ingestion( 

11 settings: Any, 

12 qdrant_manager: Any, 

13 *, 

14 project: str | None, 

15 source_type: str | None, 

16 source: str | None, 

17 force: bool, 

18 resume: bool = True, 

19 metrics_dir: str | None = None, 

20) -> None: 

21 from qdrant_loader.core.async_ingestion_pipeline import AsyncIngestionPipeline 

22 

23 pipeline = ( 

24 AsyncIngestionPipeline(settings, qdrant_manager, metrics_dir=metrics_dir) 

25 if metrics_dir 

26 else AsyncIngestionPipeline(settings, qdrant_manager) 

27 ) 

28 logger = LoggingConfig.get_logger(__name__) 

29 ingestion_error: Exception | None = None 

30 try: 

31 await pipeline.process_documents( 

32 project_id=project, 

33 source_type=source_type, 

34 source=source, 

35 force=force, 

36 resume=resume, 

37 ) 

38 except Exception as e: 

39 ingestion_error = e 

40 sanitized_traceback = sanitize_exception_message(traceback.format_exc()) 

41 logger.error( 

42 "Ingestion failed", 

43 error=sanitize_exception_message(e), 

44 error_type=type(e).__name__, 

45 sanitized_traceback=sanitized_traceback, 

46 ) 

47 cleanup_error: Exception | None = None 

48 try: 

49 await pipeline.cleanup() 

50 except Exception as e: 

51 cleanup_error = e 

52 sanitized_traceback = sanitize_exception_message(traceback.format_exc()) 

53 if ingestion_error is not None: 

54 logger.error( 

55 "Cleanup failed after ingestion exception", 

56 error=sanitize_exception_message(e), 

57 error_type=type(e).__name__, 

58 sanitized_traceback=sanitized_traceback, 

59 ) 

60 else: 

61 logger.error( 

62 "Cleanup failed after successful ingestion", 

63 error=sanitize_exception_message(e), 

64 error_type=type(e).__name__, 

65 sanitized_traceback=sanitized_traceback, 

66 ) 

67 if ingestion_error is not None: 

68 raise ingestion_error 

69 if cleanup_error is not None: 

70 raise cleanup_error