Coverage for src / augint_library / cli.py: 94%
247 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-30 20:22 +0000
1"""CLI interface for augint-library.
3This module demonstrates best practices for creating command-line interfaces
4that wrap library functions. It showcases:
5- Click framework for elegant CLI design
6- Proper separation between CLI and library logic
7- Error handling and user feedback
8- Integration with logging, telemetry, and feature flags
9- Command organization with groups
11Common Use Cases:
12 1. Creating user-friendly CLIs for Python libraries
13 2. Building multi-command tools
14 3. Adding telemetry and feature flags to CLIs
15 4. Proper error handling and user feedback
17Examples:
18 Basic CLI usage:
19 $ ai-test-script greet Alice
20 Hi Alice
22 $ ai-test-script fetch-data /api/users
23 ✓ Successfully fetched data from /api/users
25 Using with logging:
26 $ AI_TEST_SCRIPT_LOG_JSON=true ai-test-script fetch-data /api/users
27 {"timestamp": "2024-01-01T12:00:00Z", "level": "INFO", "message": "Fetching data", ...}
29 Error handling:
30 $ ai-test-script fetch-data /api/broken --timeout 0.1
31 ✗ Network timeout after 0.1s
33 Using feature flags:
34 $ ai-test-script feature-flags status
35 Feature Flags:
36 ✓ new_feature: enabled
37 ✗ experimental_api: disabled
39CLI Design Patterns:
40 1. Thin CLI layer - just parse args and call library
41 2. Consistent output format (✓/✗ symbols, colors)
42 3. Proper exit codes (0 for success, 1+ for errors)
43 4. JSON output option for scripting
44 5. Help text that includes examples
46Creating Your Own CLI:
47 >>> import click
48 >>> from your_library import process_data
49 >>>
50 >>> @click.command()
51 >>> @click.argument('input_file', type=click.Path(exists=True))
52 >>> @click.option('--format', '-f', default='json', help='Output format')
53 >>> def process(input_file, format):
54 ... '''Process a data file.'''
55 ... try:
56 ... result = process_data(input_file, format=format)
57 ... click.echo(f"✓ Processed {len(result)} records")
58 ... except Exception as e:
59 ... click.echo(f"✗ Error: {e}", err=True)
60 ... raise click.ClickException(str(e))
62Note:
63 This module demonstrates patterns that work for both simple scripts
64 and complex enterprise CLIs. The key is keeping the CLI layer thin
65 and delegating all business logic to the library modules.
66"""
68import contextlib
69import json
70import time
71from typing import Optional
73import click
75from .constants import (
76 CLI_RESILIENT_INITIAL_DELAY,
77 CLI_RESILIENT_RECOVERY_TIMEOUT,
78 CLI_SEPARATOR_LENGTH,
79 DEFAULT_API_TIMEOUT,
80 DEFAULT_FAILURE_RATE,
81 DEFAULT_FAILURE_THRESHOLD,
82 DEFAULT_RETRY_ATTEMPTS,
83)
84from .core import fetch_data, print_hi
85from .exceptions import NetworkError, ValidationError
86from .feature_flags import get_flags
87from .logging import setup_logging
88from .resilience import (
89 CircuitOpenError,
90 circuit_breaker,
91 get_circuit_breakers,
92 reset_circuit_breaker,
93 retry,
94)
96# Telemetry imports (optional - requires telemetry group dependencies)
97try:
98 from .telemetry import get_telemetry_client, track_command_execution
99except ImportError:
100 from typing import Any, Callable
102 # Create no-op decorator when telemetry is not available
103 def track_command_execution(func: Callable[..., Any]) -> Callable[..., Any]:
104 """No-op decorator when telemetry is not available."""
105 return func
107 def get_telemetry_client() -> None: # type: ignore[misc]
108 """Dummy telemetry client when telemetry is not available."""
109 raise click.ClickException(
110 "Telemetry is not available. Install with: uv sync --group telemetry"
111 )
114# CLI is not part of the library API
115__all__: list[str] = []
118@click.group()
119@click.pass_context
120def cli(ctx: click.Context) -> None:
121 """Main CLI entry point for augint-library."""
122 # Store shared context for subcommands
123 ctx.ensure_object(dict)
126@cli.command()
127@click.argument("name", default="there", type=str)
128@click.option("--json-logs", is_flag=True, help="Output logs in JSON format")
129@click.option("--log-level", default="INFO", help="Logging level (DEBUG, INFO, WARNING, ERROR)")
130@track_command_execution
131def greet(name: str, json_logs: bool, log_level: str) -> None:
132 """Greet someone by name.
134 Args:
135 name: The name to greet (default: "there").
136 json_logs: Whether to output logs in JSON format.
137 log_level: The logging level to use.
138 """
139 # Setup logging based on CLI options
140 logger = setup_logging("cli", level=log_level, json_format=json_logs)
141 logger.info("Starting greeting command", extra={"target_name": name})
143 try:
144 print_hi(name)
145 logger.info("Greeting completed successfully", extra={"target_name": name})
146 except Exception as e:
147 logger.exception(
148 "Failed to complete greeting", extra={"target_name": name, "error": str(e)}
149 )
150 raise
153# Apply resilience patterns to the library function
154resilient_fetch_data = retry(
155 max_attempts=DEFAULT_RETRY_ATTEMPTS, initial_delay=CLI_RESILIENT_INITIAL_DELAY
156)(
157 circuit_breaker(
158 failure_threshold=DEFAULT_FAILURE_THRESHOLD, recovery_timeout=CLI_RESILIENT_RECOVERY_TIMEOUT
159 )(fetch_data)
160)
163@cli.command("fetch-data")
164@click.argument("endpoint", type=str)
165@click.option("--timeout", default=DEFAULT_API_TIMEOUT, help="Request timeout in seconds")
166@click.option(
167 "--failure-rate", default=DEFAULT_FAILURE_RATE, help="Simulated failure rate (0.0-1.0)"
168)
169@click.option("--json-logs", is_flag=True, help="Output logs in JSON format")
170@click.option("--log-level", default="INFO", help="Logging level")
171@track_command_execution
172def fetch_data_command(
173 endpoint: str, timeout: float, failure_rate: float, json_logs: bool, log_level: str
174) -> None:
175 """Fetch data from a remote endpoint with resilience patterns.
177 This command demonstrates retry and circuit breaker patterns
178 for handling transient failures in external service calls.
179 """
180 logger = setup_logging("cli", level=log_level, json_format=json_logs)
181 logger.info("Fetching data", extra={"endpoint": endpoint, "timeout": timeout})
183 try:
184 result = resilient_fetch_data(endpoint, timeout=timeout, failure_rate=failure_rate)
185 click.echo(f"Success: {result['status']}")
186 click.echo(f"Data: {result['data']}")
187 logger.info("Data fetched successfully", extra={"endpoint": endpoint})
188 except CircuitOpenError as e:
189 click.echo(f"Circuit breaker OPEN: {e.message}", err=True)
190 logger.warning("Circuit breaker open", extra=e.details)
191 raise click.ClickException("Service temporarily unavailable") from e
192 except NetworkError as e:
193 click.echo(f"Network error: {e.message}", err=False)
194 logger.exception("Network request failed", extra=e.details)
195 raise click.ClickException("Failed to fetch data after retries") from e
198@cli.group("circuit-breaker")
199def circuit_breaker_group() -> None:
200 """Manage circuit breakers."""
203@circuit_breaker_group.command("status")
204def circuit_breaker_status() -> None:
205 """Show status of all circuit breakers."""
206 breakers = get_circuit_breakers()
208 if not breakers:
209 click.echo("No circuit breakers registered yet.")
210 return
212 click.echo("Circuit Breaker Status:")
213 click.echo("-" * CLI_SEPARATOR_LENGTH)
215 for name, state in breakers.items():
216 click.echo(f"Name: {name}")
217 click.echo(f" State: {state['state'].upper()}")
218 click.echo(f" Failures: {state['failure_count']}")
219 click.echo(f" Successes: {state['success_count']}")
220 if state["last_failure"]: 220 ↛ 224line 220 didn't jump to line 224 because the condition on line 220 was always true
221 click.echo(
222 f" Last failure: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(state['last_failure']))}"
223 )
224 click.echo()
227@circuit_breaker_group.command("reset")
228@click.argument("name", required=False)
229def circuit_breaker_reset(name: Optional[str]) -> None:
230 """Reset circuit breaker(s) to closed state."""
231 if name:
232 if reset_circuit_breaker(name):
233 click.echo(f"Reset circuit breaker: {name}")
234 else:
235 click.echo(f"Circuit breaker not found: {name}", err=True)
236 raise click.ClickException("Invalid circuit breaker name")
237 else:
238 # Reset all breakers
239 breakers = get_circuit_breakers()
240 for breaker_name in breakers:
241 reset_circuit_breaker(breaker_name)
242 click.echo(f"Reset {len(breakers)} circuit breaker(s)")
245@cli.group("feature-flags")
246def feature_flags_group() -> None:
247 """Manage feature flags."""
250@feature_flags_group.command("list")
251@click.option("--json", "output_json", is_flag=True, help="Output as JSON")
252def feature_flags_list(output_json: bool) -> None:
253 """List all feature flags and their current state."""
254 flags = get_flags()
255 all_flags = flags.get_all_flags()
257 if not all_flags:
258 click.echo("No feature flags registered.")
259 return
261 if output_json:
262 click.echo(json.dumps(all_flags, indent=2))
263 else:
264 click.echo("Feature Flags:")
265 click.echo("-" * CLI_SEPARATOR_LENGTH)
267 for name, info in sorted(all_flags.items()):
268 click.echo(f"Name: {name}")
269 click.echo(f" State: {info['state']}")
270 click.echo(f" Default: {info['default']}")
271 if info["description"]:
272 click.echo(f" Description: {info['description']}")
273 if info["metadata"]:
274 click.echo(f" Metadata: {info['metadata']}")
275 click.echo()
278@feature_flags_group.command("set")
279@click.argument("name")
280@click.argument("state")
281def feature_flags_set(name: str, state: str) -> None:
282 """Set the state of a feature flag.
284 STATE can be: enabled, disabled, percentage:N (0-100), or conditional.
286 Examples:
287 ai-test-script feature-flags set new_feature enabled
288 ai-test-script feature-flags set beta_feature percentage:50
289 """
290 flags = get_flags()
292 try:
293 flags.set_state(name, state)
294 click.echo(f"Set feature flag '{name}' to '{state}'")
295 except Exception as e:
296 click.echo(f"Error: {e}", err=True)
297 raise click.ClickException(str(e)) from e
300@feature_flags_group.command("check")
301@click.argument("name")
302@click.option("--user-id", help="User ID for context")
303@click.option("--environment", help="Environment for context")
304def feature_flags_check(name: str, user_id: Optional[str], environment: Optional[str]) -> None:
305 """Check if a feature flag is enabled.
307 Optionally provide context for percentage and conditional flags.
308 """
309 flags = get_flags()
311 context = {}
312 if user_id:
313 context["user_id"] = user_id
314 if environment:
315 context["environment"] = environment
317 enabled = flags.is_enabled(name, context)
319 try:
320 state = flags.get_state(name)
321 except ValidationError:
322 # Unknown flag - show as disabled
323 state = "unknown (defaulting to disabled)"
325 click.echo(f"Flag: {name}")
326 click.echo(f"State: {state}")
327 click.echo(f"Enabled: {'Yes' if enabled else 'No'}")
329 if context:
330 click.echo(f"Context: {context}")
333@feature_flags_group.command("register")
334@click.argument("name")
335@click.option("--default", default="disabled", help="Default state (default: disabled)")
336@click.option("--description", help="Description of the flag")
337def feature_flags_register(name: str, default: str, description: Optional[str]) -> None:
338 """Register a new feature flag."""
339 flags = get_flags()
341 try:
342 flags.register(name, default=default, description=description)
343 click.echo(f"Registered feature flag '{name}' with default state '{default}'")
344 except Exception as e:
345 click.echo(f"Error: {e}", err=True)
346 raise click.ClickException(str(e)) from e
349@cli.group("telemetry")
350def telemetry_group() -> None:
351 """Manage anonymous usage telemetry settings.
353 Telemetry helps improve augint-library by collecting anonymous usage
354 statistics and error reports. No personal information is ever collected.
355 """
356 # Check if telemetry is available at group level
357 with contextlib.suppress(click.ClickException):
358 get_telemetry_client()
361@telemetry_group.command("status")
362def telemetry_status() -> None:
363 """Show current telemetry status and configuration."""
364 client = get_telemetry_client()
366 click.echo("Telemetry Status")
367 click.echo("-" * 40)
368 click.echo(f"Enabled: {'Yes' if client.enabled else 'No'}")
370 if client.enabled:
371 # Show anonymous ID
372 anonymous_id = client._get_anonymous_id()
373 click.echo(f"Anonymous ID: {anonymous_id}")
374 click.echo()
375 click.echo("Data collected:")
376 click.echo("- Command usage (names only, no arguments)")
377 click.echo("- Success/failure rates")
378 click.echo("- Performance metrics")
379 click.echo("- Python and OS versions")
380 click.echo("- Anonymous error reports")
381 click.echo()
382 click.echo("Data NOT collected:")
383 click.echo("- Personal information")
384 click.echo("- File paths or contents")
385 click.echo("- Command arguments")
386 click.echo("- Network locations")
387 else:
388 click.echo()
389 click.echo("Telemetry is disabled. To help improve augint-library, run:")
390 click.echo(" ai-test-script telemetry enable")
393@telemetry_group.command("enable")
394@click.option("--yes", "-y", is_flag=True, help="Skip confirmation prompt")
395def telemetry_enable(yes: bool) -> None:
396 """Enable anonymous telemetry."""
397 client = get_telemetry_client()
399 if client.enabled:
400 click.echo("Telemetry is already enabled.")
401 return
403 if not yes:
404 click.echo("augint-library Telemetry")
405 click.echo("-" * 40)
406 click.echo()
407 click.echo("Help improve augint-library by sharing anonymous usage data.")
408 click.echo()
409 click.echo("What we collect:")
410 click.echo(" ✓ Command usage patterns (no arguments)")
411 click.echo(" ✓ Error types and frequencies")
412 click.echo(" ✓ Performance metrics")
413 click.echo(" ✓ Python/OS versions")
414 click.echo()
415 click.echo("What we DON'T collect:")
416 click.echo(" ✗ Personal information")
417 click.echo(" ✗ File paths or contents")
418 click.echo(" ✗ IP addresses or hostnames")
419 click.echo(" ✗ Command arguments or data")
420 click.echo()
422 if not click.confirm("Enable telemetry?"):
423 click.echo("Telemetry remains disabled.")
424 return
426 client.set_consent(enabled=True)
427 click.echo("✓ Telemetry enabled. Thank you for helping improve augint-library!")
428 click.echo(" You can disable this at any time with: ai-test-script telemetry disable")
431@telemetry_group.command("disable")
432def telemetry_disable() -> None:
433 """Disable telemetry."""
434 client = get_telemetry_client()
436 if not client.enabled:
437 click.echo("Telemetry is already disabled.")
438 return
440 client.set_consent(enabled=False)
441 click.echo("✓ Telemetry disabled.")
442 click.echo(" Your anonymous ID has been preserved for consistency if you re-enable.")
445@telemetry_group.command("test")
446def telemetry_test() -> None:
447 """Send a test event to verify telemetry is working."""
448 client = get_telemetry_client()
450 if not client.enabled:
451 click.echo("Telemetry is disabled. Enable it first with:")
452 click.echo(" ai-test-script telemetry enable")
453 return
455 click.echo("Sending test telemetry event...")
457 # Track test command
458 client.track_command("telemetry_test", success=True, duration=0.1)
460 # Track test metric
461 client.track_metric("test_metric", 42.0, unit="test", tags={"type": "verification"})
463 # Track test error
464 def _create_test_error() -> None:
465 raise ValueError("This is a test error for telemetry verification")
467 try:
468 _create_test_error()
469 except ValueError as e:
470 client.track_error(e, {"context": "telemetry_test_command"})
472 # Flush to ensure events are sent
473 client.flush()
475 click.echo("✓ Test events sent successfully!")
476 click.echo(" Check your Sentry dashboard to verify they were received.")
479if __name__ == "__main__":
480 cli()