Coverage for queue_cli.py: 0.00%

1#!/usr/bin/env python3 

2""" 

3Queue CLI - Interactive tool for testing RabbitMQ/Celery queues. 

4 

5A comprehensive CLI tool for: 

6- Sending messages to RabbitMQ queues 

7- Managing message templates 

8- Monitoring task execution 

9- Learning about distributed messaging systems 

10 

11Examples: 

12 # Send a message using a template 

13 uv run python queue_cli.py send --template completed-match 

14 

15 # List available templates 

16 uv run python queue_cli.py templates 

17 

18 # Show specific template 

19 uv run python queue_cli.py templates --show tbd-match 

20 

21 # Send with debug output 

22 uv run python queue_cli.py send --template minimal --debug 

23 

24 # Use direct RabbitMQ (instead of Celery) 

25 uv run python queue_cli.py send --template scheduled-match --rabbitmq 

26""" 

27 

28import typer 

29from rich.console import Console 

30 

31from queue_cli.commands.send import send_message 

32from queue_cli.commands.templates import templates_command 

33 

34# Create Typer app 

35app = typer.Typer( 

36 name="queue-cli", 

37 help="Interactive CLI tool for testing RabbitMQ/Celery queues", 

38 add_completion=False, 

39 rich_markup_mode="rich", 

40) 

41 

42console = Console() 

43 

44 

45@app.command("send") 

46def send( 

47 template: str = typer.Option(None, "--template", "-t", help="Template name to use"), 

48 file: str = typer.Option(None, "--file", "-f", help="JSON file to send"), 

49 json_str: str = typer.Option(None, "--json", "-j", help="JSON string to send"), 

50 queue: str = typer.Option(None, "--queue", "-q", help="Target queue name"), 

51 use_celery: bool = typer.Option(True, "--celery/--rabbitmq", help="Use Celery task vs direct RabbitMQ publish"), 

52 debug: bool = typer.Option(False, "--debug", "-d", help="Enable debug output"), 

53) -> None: 

54 """ 

55 Send a message to the queue. 

56 

57 This is the primary command for testing queue functionality. Messages can be: 

58 - Loaded from built-in templates (--template) 

59 - Read from JSON files (--file) 

60 - Passed directly as JSON (--json) 

61 

62 By default, messages are sent via Celery tasks (--celery), which is the 

63 production method. Use --rabbitmq to publish directly to RabbitMQ for 

64 educational purposes and to see the low-level mechanics. 

65 """ 

66 from pathlib import Path 

67 

68 send_message( 

69 template=template, 

70 file=Path(file) if file else None, 

71 json_str=json_str, 

72 queue=queue, 

73 use_celery=use_celery, 

74 debug=debug, 

75 ) 

76 

77 

78@app.command("templates") 

79def templates( 

80 show: str = typer.Option(None, "--show", "-s", help="Show specific template"), 

81) -> None: 

82 """ 

83 Manage message templates. 

84 

85 Templates are pre-configured message payloads for common scenarios: 

86 - completed-match: Match with final scores 

87 - scheduled-match: Upcoming match without scores 

88 - tbd-match: Played match, awaiting scores 

89 - minimal: Required fields only 

90 

91 List all templates or show a specific one with --show. 

92 """ 

93 templates_command(show=show) 

94 

95 

96@app.callback() 

97def main() -> None: 

98 """ 

99 Queue CLI - Interactive tool for testing RabbitMQ/Celery queues. 

100 

101 This tool helps you learn about distributed messaging systems by: 

102 1. Sending messages to RabbitMQ queues 

103 2. Monitoring Celery task execution 

104 3. Understanding message schemas and validation 

105 4. Visualizing the message flow 

106 

107 Phase 1 commands: 

108 - send: Send messages to the queue 

109 - templates: Manage message templates 

110 

111 Coming soon (Phase 2+): 

112 - status: Check task status 

113 - queues: View queue statistics 

114 - workers: Monitor worker activity 

115 - watch: Real-time queue monitoring 

116 """ 

117 pass 

118 

119 

120def version_callback(value: bool) -> None: 

121 """Print version and exit.""" 

122 if value: 

123 console.print("Queue CLI v0.1.0 - Phase 1") 

124 raise typer.Exit() 

125 

126 

127@app.command() 

128def version( 

129 version_flag: bool = typer.Option( 

130 None, 

131 "--version", 

132 "-v", 

133 callback=version_callback, 

134 is_eager=True, 

135 help="Show version", 

136 ), 

137) -> None: 

138 """Show version information.""" 

139 pass 

140 

141 

142if __name__ == "__main__": 

143 app()