Coverage for queue_cli/utils/display.py: 0.00%

1""" 

2Rich display utilities for beautiful CLI output. 

3""" 

4 

5import json 

6from typing import Any 

7 

8from rich.console import Console 

9from rich.json import JSON 

10from rich.panel import Panel 

11from rich.table import Table 

12 

13console = Console() 

14 

15 

16def print_header(title: str) -> None: 

17 """Print a formatted header.""" 

18 console.print() 

19 console.print(Panel(f"[bold cyan]{title}[/bold cyan]", expand=False)) 

20 console.print() 

21 

22 

23def print_section(title: str, emoji: str = "📋") -> None: 

24 """Print a section header.""" 

25 console.print(f"\n{emoji} [bold]{title}[/bold]") 

26 

27 

28def print_success(message: str) -> None: 

29 """Print a success message.""" 

30 console.print(f" [green]✓[/green] {message}") 

31 

32 

33def print_error(message: str) -> None: 

34 """Print an error message.""" 

35 console.print(f" [red]✗[/red] {message}") 

36 

37 

38def print_warning(message: str) -> None: 

39 """Print a warning message.""" 

40 console.print(f" [yellow]⚠[/yellow] {message}") 

41 

42 

43def print_info(message: str, bullet: str = "•") -> None: 

44 """Print an info message.""" 

45 console.print(f" {bullet} {message}") 

46 

47 

48def print_json(data: dict[str, Any], title: str | None = None) -> None: 

49 """Print formatted JSON.""" 

50 if title: 

51 print_section(title, "📨") 

52 

53 json_str = json.dumps(data, indent=2) 

54 console.print(JSON(json_str)) 

55 

56 

57def print_key_value(key: str, value: Any, indent: int = 1) -> None: 

58 """Print a key-value pair.""" 

59 indent_str = " " * indent 

60 console.print(f"{indent_str}[cyan]{key}:[/cyan] {value}") 

61 

62 

63def print_connection_info(broker_url: str, is_connected: bool = False) -> None: 

64 """Print connection information.""" 

65 print_section("RabbitMQ Connection", "🔌") 

66 

67 status = "[green]Established[/green]" if is_connected else "[yellow]Connecting...[/yellow]" 

68 

69 print_info(f"Broker: {broker_url}") 

70 print_info(f"Status: {status}") 

71 

72 

73def print_publish_info( 

74 queue: str, 

75 exchange: str, 

76 routing_key: str, 

77 message_count: int, 

78 consumer_count: int, 

79) -> None: 

80 """Print message publish information.""" 

81 print_section("Message Published", "📤") 

82 

83 print_info(f"Queue: [bold]{queue}[/bold]") 

84 print_info(f"Exchange: {exchange}") 

85 print_info(f"Routing Key: {routing_key}") 

86 print_info(f"Messages in Queue: {message_count}") 

87 print_info(f"Active Consumers: {consumer_count}") 

88 

89 

90def print_validation_result(is_valid: bool, errors: list[str], warnings: list[str]) -> None: 

91 """Print validation results.""" 

92 print_section("Message Validation", "📋") 

93 

94 if is_valid: 

95 print_success("Schema validation passed") 

96 else: 

97 print_error("Schema validation failed") 

98 

99 for error in errors: 

100 print_error(error) 

101 

102 for warning in warnings: 

103 print_warning(warning) 

104 

105 

106def print_task_info(task_id: str, task_name: str, state: str) -> None: 

107 """Print Celery task information.""" 

108 print_section("Celery Task", "⚙️") 

109 

110 print_info(f"Task ID: [bold]{task_id}[/bold]") 

111 print_info(f"Task Name: {task_name}") 

112 print_info(f"State: [bold]{state}[/bold]") 

113 

114 

115def print_next_steps(task_id: str, queue: str) -> None: 

116 """Print next steps for the user.""" 

117 console.print() 

118 console.print("[bold]🔍 Next Steps:[/bold]") 

119 console.print(f" • Check task status: [cyan]uv run python queue_cli.py status {task_id}[/cyan]") 

120 console.print(f" • View queue: [cyan]uv run python queue_cli.py queues --name {queue}[/cyan]") 

121 console.print(" • View workers: [cyan]uv run python queue_cli.py workers[/cyan]") 

122 console.print() 

123 

124 

125def print_template_list(templates: list[str]) -> None: 

126 """Print list of available templates.""" 

127 print_header("Available Templates") 

128 

129 table = Table(show_header=True, header_style="bold cyan") 

130 table.add_column("Template Name", style="cyan") 

131 table.add_column("Description") 

132 

133 template_descriptions = { 

134 "completed-match": "Match with final scores", 

135 "scheduled-match": "Upcoming match without scores", 

136 "tbd-match": "Played match, awaiting scores", 

137 "minimal": "Required fields only", 

138 } 

139 

140 for template in sorted(templates): 

141 desc = template_descriptions.get(template, "Custom template") 

142 table.add_row(template, desc) 

143 

144 console.print(table) 

145 console.print() 

146 

147 

148def print_queue_stats(stats: dict[str, Any]) -> None: 

149 """Print queue statistics.""" 

150 print_header("Queue Statistics") 

151 

152 table = Table(show_header=True, header_style="bold cyan") 

153 table.add_column("Queue", style="cyan") 

154 table.add_column("Messages", justify="right") 

155 table.add_column("Consumers", justify="right") 

156 

157 table.add_row( 

158 stats.get("queue_name", "N/A"), 

159 str(stats.get("message_count", 0)), 

160 str(stats.get("consumer_count", 0)), 

161 ) 

162 

163 console.print(table) 

164 console.print() 

165 

166 

167def print_debug_mode() -> None: 

168 """Print debug mode indicator.""" 

169 console.print("[yellow]🐛 Debug mode enabled[/yellow]") 

170 

171 

172def print_divider() -> None: 

173 """Print a visual divider.""" 

174 console.print("[dim]" + "─" * 80 + "[/dim]")