#!/usr/bin/env python3 """ Common Expression Language (CEL) Command Line Interface A powerful CLI for evaluating CEL expressions with support for: - Interactive REPL mode with history and syntax highlighting - File-based expression evaluation - JSON context input/output - Batch processing - Performance timing - Beautiful output formatting """ import json import sys import time from importlib.metadata import PackageNotFoundError, version from pathlib import Path from typing import Any, Dict, Optional, Tuple import typer # Prompt toolkit imports for enhanced REPL from prompt_toolkit import PromptSession from prompt_toolkit.auto_suggest import AutoSuggestFromHistory from prompt_toolkit.completion import WordCompleter from prompt_toolkit.history import FileHistory from prompt_toolkit.lexers import PygmentsLexer from prompt_toolkit.styles import Style from pygments import token # Pygments imports for syntax highlighting from pygments.lexer import RegexLexer from rich.console import Console from rich.panel import Panel from rich.syntax import Syntax from rich.table import Table from typing_extensions import Annotated # Import directly from relative modules to avoid circular imports from .cel import Context, evaluate from .stdlib import STDLIB_FUNCTIONS, add_stdlib_to_context # Initialize Rich console console = Console() def get_version() -> str: """Get the version of the CEL package.""" try: return version("common-expression-language") except PackageNotFoundError: return "unknown (development)" def version_callback(value: bool): """Print version and exit.""" if value: console.print(f"cel {get_version()}") raise typer.Exit() class CELLexer(RegexLexer): """Custom Pygments lexer for CEL syntax highlighting in the REPL.""" name = "CEL" aliases = ["cel"] filenames = ["*.cel"] tokens = { "root": [ # Keywords and constants (r"\b(true|false|null)\b", token.Keyword.Constant), (r"\b(in|if|else|and|or|not)\b", token.Keyword), # Built-in functions ( r"\b(size|has|timestamp|duration|int|uint|double|string|bytes|bool|dyn|type|" r"startsWith|endsWith|contains|matches|substring|charAt|indexOf|lastIndexOf|" r"replace|split|join|lowerAscii|upperAscii|trim|reverse|min|max|" r"distinct|flatten|slice|sort|first|last)\b(?=\()", token.Name.Function, ), # String literals (r'"([^"\\]|\\.)*"', token.String.Double), (r"'([^'\\]|\\.)*'", token.String.Single), # Byte literals (r'b"([^"\\]|\\.)*"', token.String.Affix), (r"b'([^'\\]|\\.)*'", token.String.Affix), # Numeric literals (r"\b[0-9]+\.[0-9]*([eE][+-]?[0-9]+)?\b", token.Number.Float), (r"\b[0-9]+[eE][+-]?[0-9]+\b", token.Number.Float), (r"\b[0-9]+u\b", token.Number.Integer), # unsigned integers (r"\b[0-9]+\b", token.Number.Integer), # Operators (r"[+\-*/%]", token.Operator), (r"[<>=!]=?", token.Operator.Comparison), (r"&&|\|\||!", token.Operator.Logical), (r"\?|:", token.Operator.Conditional), # Punctuation (r"[\[\]{}().,]", token.Punctuation), # Identifiers (variables, fields) (r"\b[a-zA-Z_][a-zA-Z0-9_]*\b", token.Name.Variable), # Whitespace (r"\s+", token.Whitespace), # Comments (if we want to support them in REPL) (r"#.*$", token.Comment.Single), ], } # CLI application app = typer.Typer( name="cel", help="Common Expression Language (CEL) Command Line Interface", add_completion=False, rich_markup_mode="rich", context_settings={"help_option_names": ["-h", "--help"]}, ) class CELFormatter: """Enhanced formatter using Rich for beautiful output.""" @staticmethod def display(console_obj: Console, result: Any, format_type: str = "auto") -> None: """Main entry point for displaying results - uses Rich renderables for efficiency.""" rich_renderable = CELFormatter.get_rich_renderable(result, format_type) console_obj.print(rich_renderable) @staticmethod def get_rich_renderable(result: Any, format_type: str = "auto") -> Any: """Returns a Rich renderable object for the given result and format type.""" if format_type == "json": json_str = json.dumps(result, indent=2, default=str, ensure_ascii=False) return Syntax(json_str, "json", theme="monokai", line_numbers=False) elif format_type == "pretty": return CELFormatter._get_pretty_renderable(result) elif format_type == "python": return repr(result) else: # auto return CELFormatter._get_auto_renderable(result) @staticmethod def format_result(result: Any, format_type: str = "auto") -> str: """Format result as string (for backward compatibility and testing).""" rich_renderable = CELFormatter.get_rich_renderable(result, format_type) # For Rich objects, we need to capture their output if hasattr(rich_renderable, "__rich__") or hasattr(rich_renderable, "__rich_console__"): with console.capture() as capture: console.print(rich_renderable) return capture.get() return str(rich_renderable) @staticmethod def _get_pretty_renderable(result: Any) -> Any: """Get Rich renderable for pretty-formatted result.""" if isinstance(result, dict): table = Table(title="Dictionary Result", show_header=True, header_style="bold magenta") table.add_column("Key", style="cyan") table.add_column("Value", style="green") for k, v in result.items(): table.add_row(str(k), str(v)) return table elif isinstance(result, list): table = Table(title="List Result", show_header=True, header_style="bold magenta") table.add_column("Index", style="cyan") table.add_column("Value", style="green") for i, v in enumerate(result): table.add_row(str(i), str(v)) return table else: # Use f-string as suggested type_name = type(result).__name__ return f"{result} ({type_name})" @staticmethod def _get_auto_renderable(result: Any) -> Any: """Get Rich renderable for auto-formatted result.""" if isinstance(result, (dict, list)) and len(str(result)) > 100: return CELFormatter._get_pretty_renderable(result) return str(result) class CELEvaluator: """Enhanced CEL expression evaluator.""" def __init__(self, context: Optional[Dict[str, Any]] = None): """Initialize evaluator with optional context.""" self.context = context or {} self._cel_context = None self._update_cel_context() def _update_cel_context(self): """Update the internal CEL context object.""" if self.context: self._cel_context = Context(self.context) else: self._cel_context = Context() # Always add stdlib functions to the context add_stdlib_to_context(self._cel_context) def evaluate(self, expression: str) -> Any: """Evaluate a CEL expression.""" if not expression.strip(): raise ValueError("Empty expression") return evaluate(expression, self._cel_context) def update_context(self, new_context: Dict[str, Any]): """Update the evaluation context.""" self.context.update(new_context) self._update_cel_context() def get_context_vars(self) -> Dict[str, Any]: """Get current context variables for display.""" return self.context.copy() class InteractiveCELREPL: """Enhanced REPL with prompt_toolkit features.""" def __init__(self, evaluator: CELEvaluator, history_limit: int = 10): """Initialize REPL with enhanced features.""" self.evaluator = evaluator self.history: list[Tuple[str, Any]] = [] self.history_limit = history_limit # CEL keywords and functions for completion - stored as instance variables self.cel_keywords = [ "true", "false", "null", "if", "else", "in", "and", "or", "not", ] self.cel_functions = [ "size", "has", "timestamp", "duration", "int", "uint", "double", "string", "bytes", # Extended standard-library functions (registered by the CLI). # Simple names only; namespaced helpers (math.*, sets.*, base64.*, # lists.*) are surfaced via the namespace prefixes below. *sorted(name for name in STDLIB_FUNCTIONS if "." not in name), "math", "sets", "base64", "lists", "strings", ] # Command dispatch dictionary for cleaner organization self.commands = { "help": self._show_help, "history": self._show_history, } # Setup prompt session with history, autocompletion, and syntax highlighting history_file = Path.home() / ".cel_history" self.session: PromptSession[str] = PromptSession( history=FileHistory(str(history_file)), auto_suggest=AutoSuggestFromHistory(), complete_while_typing=True, lexer=PygmentsLexer(CELLexer), # Add real-time syntax highlighting ) self._update_completer() # Rich styling for the REPL self.style = Style.from_dict( { "prompt": "#00aa00 bold", "result": "#0088ff", "error": "#ff0066", } ) def run(self): """Run the enhanced REPL.""" console.print( Panel.fit( "[bold green]CEL Interactive REPL[/bold green]\n" "Enhanced with history, autocompletion, and syntax highlighting\n" "Type 'help' for commands, 'exit' to quit", border_style="green", ) ) while True: try: # Get input with enhanced prompt expression = self.session.prompt("cel> ", style=self.style) if not expression.strip(): continue # Handle REPL commands command_parts = expression.strip().split() command = command_parts[0].lower() if command_parts else "" if command in ["exit", "quit"]: console.print("[yellow]Goodbye![/yellow]") break elif command == "context": if len(command_parts) > 1: # Setting context: context {"key": "value"} context_json = " ".join(command_parts[1:]) self._set_context(context_json) else: # Showing context: context self._show_context() continue elif command in self.commands: self.commands[command]() continue elif command == "load" and len(command_parts) > 1: filename = " ".join(command_parts[1:]) # Handle filenames with spaces self._load_context(filename) continue # Evaluate expression start_time = time.time() result = self.evaluator.evaluate(expression) eval_time = time.time() - start_time # Display result using streamlined formatter CELFormatter.display(console, result, "pretty") # Show timing console.print(f"[dim]Evaluated in {eval_time * 1000:.2f}ms[/dim]") # Add to history (keep limited) self.history.append((expression, result)) if len(self.history) > 100: # Keep last 100 items self.history = self.history[-100:] except KeyboardInterrupt: console.print("\n[yellow]Goodbye![/yellow]") break except EOFError: console.print("\n[yellow]Goodbye![/yellow]") break except Exception as e: console.print(f"[red]Error: {e}[/red]") def _show_help(self): """Show enhanced REPL help.""" help_table = Table(title="REPL Commands", show_header=True, header_style="bold magenta") help_table.add_column("Command", style="cyan") help_table.add_column("Description", style="green") help_table.add_row("help", "Show this help message") help_table.add_row("context", "Show current context variables") help_table.add_row("context ", "Set context variables from JSON") help_table.add_row("history", "Show expression history") help_table.add_row("load ", "Load JSON context from file") help_table.add_row("exit/quit", "Exit the REPL") help_table.add_row("Ctrl-C", "Exit the REPL") console.print(help_table) console.print("\n[bold]CEL Examples:[/bold]") examples = [ "1 + 2", '"hello" + " world"', "[1, 2, 3].size()", "timestamp('2024-01-01T00:00:00Z')", 'age > 21 ? "adult" : "minor"', ] for example in examples: console.print(f" [dim]cel>[/dim] [cyan]{example}[/cyan]") def _show_context(self): """Show current context variables with Rich formatting.""" context_vars = self.evaluator.get_context_vars() if not context_vars: console.print("[dim]No context variables set[/dim]") return table = Table(title="Context Variables", show_header=True, header_style="bold magenta") table.add_column("Variable", style="cyan") table.add_column("Type", style="yellow") table.add_column("Value", style="green") for name, value in context_vars.items(): table.add_row(name, type(value).__name__, str(value)) console.print(table) def _set_context(self, context_json: str): """Set context variables from JSON string.""" try: new_context = json.loads(context_json) if not isinstance(new_context, dict): console.print("[red]Error: Context must be a JSON object (dictionary)[/red]") return # Update the context self.evaluator.update_context(new_context) # Update completer with new context variables self._update_completer() # Show what was updated context_keys = list(new_context.keys()) if len(context_keys) == 1: console.print(f"[green]Context updated: {context_keys[0]}[/green]") elif len(context_keys) <= 3: console.print(f"[green]Context updated: {', '.join(context_keys)}[/green]") else: console.print(f"[green]Context updated: {len(context_keys)} variables[/green]") except json.JSONDecodeError as e: console.print(f"[red]Error: Invalid JSON - {e}[/red]") except Exception as e: console.print(f"[red]Error updating context: {e}[/red]") def _show_history(self): """Show expression history with Rich formatting.""" if not self.history: console.print("[dim]No history available[/dim]") return table = Table(title="Expression History", show_header=True, header_style="bold magenta") table.add_column("#", style="dim", width=3) table.add_column("Expression", style="cyan") table.add_column("Result", style="green") # Show last N items based on history_limit recent_history = self.history[-self.history_limit :] for i, (expr, result) in enumerate(recent_history, 1): # Truncate long results result_str = str(result) if len(result_str) > 50: result_str = result_str[:47] + "..." table.add_row(str(i), expr, result_str) console.print(table) def _update_completer(self): """Update the completer with current context variables.""" words = self.cel_keywords + self.cel_functions + list(self.evaluator.context.keys()) self.session.completer = WordCompleter(words) def _load_context(self, filename: str): """Load context from JSON file.""" try: with open(filename, "r") as f: context = json.load(f) self.evaluator.update_context(context) console.print(f"[green]Loaded context from {filename}[/green]") # Update completer with new context variables self._update_completer() except Exception as e: console.print(f"[red]Error loading context: {e}[/red]") def load_context_from_file(filename: Path) -> Dict[str, Any]: """Load context from JSON file with Rich error handling.""" try: with open(filename, "r") as f: return json.load(f) except json.JSONDecodeError as e: console.print(f"[red]Error: Invalid JSON in {filename}: {e}[/red]") raise typer.Exit(1) from e except FileNotFoundError as e: console.print(f"[red]Error: Context file '{filename}' not found[/red]") raise typer.Exit(1) from e def evaluate_expressions_from_file( filename: Path, evaluator: CELEvaluator, output_format: str ) -> None: """Evaluate expressions from a file with Rich output.""" try: with open(filename, "r") as f: expressions = [ line.strip() for line in f if line.strip() and not line.strip().startswith("#") ] except FileNotFoundError as e: console.print(f"[red]Error: Expression file '{filename}' not found[/red]") raise typer.Exit(1) from e if not expressions: console.print("[yellow]No expressions found in file[/yellow]") return results = [] with console.status(f"[bold green]Evaluating {len(expressions)} expressions..."): for i, expression in enumerate(expressions, 1): try: start_time = time.time() result = evaluator.evaluate(expression) eval_time = time.time() - start_time results.append( { "expression": expression, "result": result, "time_ms": eval_time * 1000, } ) except Exception as e: console.print(f"[red]Error in expression {i} '{expression}': {e}[/red]") results.append({"expression": expression, "error": str(e)}) # Display results if output_format == "json": json_output = json.dumps(results, indent=2, default=str) syntax = Syntax(json_output, "json", theme="monokai") console.print(syntax) else: table = Table(title="Expression Results", show_header=True, header_style="bold magenta") table.add_column("#", style="dim", width=3) table.add_column("Expression", style="cyan") table.add_column("Result", style="green") table.add_column("Time (ms)", style="yellow") for i, result in enumerate(results, 1): if "error" in result: table.add_row( str(i), result["expression"], f"[red]Error: {result['error']}[/red]", "—", ) else: result_str = str(result["result"]) if len(result_str) > 50: result_str = result_str[:47] + "..." table.add_row(str(i), result["expression"], result_str, f"{result['time_ms']:.2f}") console.print(table) @app.command() def main( expression: Annotated[Optional[str], typer.Argument(help="CEL expression to evaluate")] = None, context: Annotated[ Optional[str], typer.Option("-c", "--context", help="Context as JSON string") ] = None, context_file: Annotated[ Optional[Path], typer.Option("-f", "--context-file", help="Load context from JSON file"), ] = None, file: Annotated[ Optional[Path], typer.Option("--file", help="Read expressions from file (one per line)"), ] = None, output: Annotated[str, typer.Option("-o", "--output", help="Output format")] = "auto", interactive: Annotated[ bool, typer.Option("-i", "--interactive", help="Start interactive REPL mode") ] = False, timing: Annotated[bool, typer.Option("-t", "--timing", help="Show evaluation timing")] = False, verbose: Annotated[bool, typer.Option("-v", "--verbose", help="Verbose output")] = False, version: Annotated[ Optional[bool], typer.Option( "--version", callback=version_callback, is_eager=True, help="Show version and exit" ), ] = None, ): """ Evaluate CEL expressions with enhanced CLI experience. Examples: # Evaluate a simple expression cel '1 + 2' # Use context variables cel 'age > 21' --context '{"age": 25}' # Load context from file cel 'user.name' --context-file context.json # Interactive REPL mode cel --interactive # Evaluate expressions from file cel --file expressions.cel --output json # Strict CEL mode (type safety) cel '1 + 2.5' # This will fail due to mixed types cel '1.0 + 2.5' # This will work """ # Load context eval_context = {} if context: try: eval_context = json.loads(context) except json.JSONDecodeError as e: console.print(f"[red]Error: Invalid JSON in context: {e}[/red]") raise typer.Exit(1) from e if context_file: file_context = load_context_from_file(context_file) eval_context.update(file_context) # Initialize evaluator evaluator = CELEvaluator(eval_context) # Interactive mode if interactive: repl = InteractiveCELREPL(evaluator) repl.run() return # File mode if file: evaluate_expressions_from_file(file, evaluator, output) return # Single expression evaluation if not expression: console.print( "[red]Error: No expression provided. Use -i for interactive mode or provide an expression.[/red]" ) console.print("\nUse [bold]cel --help[/bold] for more information.") raise typer.Exit(1) try: # Evaluate expression start_time = time.time() result = evaluator.evaluate(expression) eval_time = time.time() - start_time # Format and output result using streamlined formatter CELFormatter.display(console, result, output) # Show timing if requested if timing or verbose: console.print(f"[dim]Evaluated in {eval_time * 1000:.2f}ms[/dim]") # Verbose output if verbose: console.print(f"[dim]Expression: {expression}[/dim]") console.print(f"[dim]Result type: {type(result).__name__}[/dim]") if eval_context: console.print(f"[dim]Context variables: {len(eval_context)}[/dim]") except Exception as e: console.print(f"[red]Error: {e}[/red]") raise typer.Exit(1) from e def cli_entry(): """Entry point for the CLI.""" app() if __name__ == "__main__": cli_entry()