Coverage for src/qdrant_loader/cli/commands/project/status_cmd.py: 84%

1from __future__ import annotations 

2 

3import json 

4from collections.abc import Mapping 

5from typing import Any 

6 

7 

8async def run_project_status( 

9 settings: Any, 

10 project_manager: Any, 

11 state_manager: Any, 

12 *, 

13 project_id: str | None, 

14 output_format: str, 

15) -> str: 

16 from rich.console import Console 

17 from rich.panel import Panel 

18 

19 async def _get_document_count(pid: str) -> int: 

20 try: 

21 return await state_manager.get_project_document_count(pid) 

22 except Exception: 

23 return 0 

24 

25 async def _get_latest_ingestion(pid: str) -> str | None: 

26 try: 

27 return await state_manager.get_project_latest_ingestion(pid) 

28 except Exception: 

29 return None 

30 

31 if project_id: 

32 from click.exceptions import BadParameter 

33 

34 context = project_manager.get_project_context(project_id) 

35 if context is None: 

36 raise BadParameter(f"Project not found: {project_id}") 

37 contexts = {project_id: context} 

38 else: 

39 contexts = project_manager.get_all_project_contexts() 

40 

41 results = [] 

42 for context in contexts.values(): 

43 if not context: 

44 continue 

45 sources = context.config.sources if context.config else None 

46 # Safely sum lengths of available source collections; treat missing/None as empty 

47 if sources: 

48 names = ("publicdocs", "git", "confluence", "jira", "localfile") 

49 counts: list[int] = [] 

50 for name in names: 

51 value = getattr(sources, name, None) 

52 if value is None: 

53 counts.append(0) 

54 continue 

55 if isinstance(value, Mapping): 

56 counts.append(len(value)) 

57 continue 

58 # Any sequence/collection or object with __len__ 

59 try: 

60 counts.append(len(value)) 

61 except Exception: 

62 counts.append(0) 

63 source_count = sum(counts) 

64 else: 

65 source_count = 0 

66 document_count = await _get_document_count(context.project_id) 

67 latest_ingestion = await _get_latest_ingestion(context.project_id) 

68 results.append( 

69 { 

70 "project_id": context.project_id, 

71 "display_name": context.display_name or "N/A", 

72 "collection_name": context.collection_name or "N/A", 

73 "source_count": source_count, 

74 "document_count": document_count, 

75 "latest_ingestion": latest_ingestion, 

76 } 

77 ) 

78 

79 if output_format.lower() == "json": 

80 return json.dumps(results, indent=2) 

81 

82 console = Console(record=True) 

83 for item in results: 

84 latest_ingestion_display = item["latest_ingestion"] or "Never" 

85 project_info = ( 

86 f"[bold cyan]Project ID:[/bold cyan] {item['project_id']}\n" 

87 f"[bold magenta]Display Name:[/bold magenta] {item['display_name']}\n" 

88 f"[bold blue]Collection:[/bold blue] {item['collection_name']}\n" 

89 f"[bold yellow]Sources:[/bold yellow] {item['source_count']}\n" 

90 f"[bold red]Documents:[/bold red] {item['document_count']}\n" 

91 f"[bold red]Latest Ingestion:[/bold red] {latest_ingestion_display}" 

92 ) 

93 console.print(Panel(project_info, title=f"Project: {item['project_id']}")) 

94 return console.export_text()