420 lines
12 KiB
Python
420 lines
12 KiB
Python
"""
|
|
pyserve config - Configuration management commands
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
import click
|
|
import yaml
|
|
|
|
|
|
@click.group("config")
|
|
def config_cmd():
|
|
"""
|
|
Configuration management commands.
|
|
|
|
\b
|
|
Commands:
|
|
validate Validate configuration file
|
|
show Display current configuration
|
|
get Get a specific configuration value
|
|
set Set a configuration value
|
|
diff Compare two configuration files
|
|
"""
|
|
pass
|
|
|
|
|
|
@config_cmd.command("validate")
|
|
@click.option(
|
|
"-c",
|
|
"--config",
|
|
"config_file",
|
|
default=None,
|
|
help="Path to configuration file",
|
|
)
|
|
@click.option(
|
|
"--strict",
|
|
is_flag=True,
|
|
help="Enable strict validation (warn on unknown fields)",
|
|
)
|
|
@click.pass_obj
|
|
def validate_cmd(ctx, config_file: Optional[str], strict: bool):
|
|
"""
|
|
Validate a configuration file.
|
|
|
|
Checks for syntax errors, missing required fields, and invalid values.
|
|
|
|
\b
|
|
Examples:
|
|
pyserve config validate
|
|
pyserve config validate -c production.yaml
|
|
pyserve config validate --strict
|
|
"""
|
|
from ..output import console, print_error, print_success, print_warning
|
|
|
|
config_path = Path(config_file or ctx.config_file)
|
|
|
|
if not config_path.exists():
|
|
print_error(f"Configuration file not found: {config_path}")
|
|
raise click.Abort()
|
|
|
|
console.print(f"Validating [cyan]{config_path}[/cyan]...")
|
|
|
|
try:
|
|
with open(config_path) as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
if data is None:
|
|
print_error("Configuration file is empty")
|
|
raise click.Abort()
|
|
|
|
from ...config import Config
|
|
|
|
config = Config.from_yaml(str(config_path))
|
|
|
|
errors = []
|
|
warnings = []
|
|
|
|
if not (1 <= config.server.port <= 65535):
|
|
errors.append(f"Invalid server port: {config.server.port}")
|
|
|
|
valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
|
|
if config.logging.level.upper() not in valid_levels:
|
|
errors.append(f"Invalid logging level: {config.logging.level}")
|
|
|
|
if config.ssl.enabled:
|
|
if not Path(config.ssl.cert_file).exists():
|
|
warnings.append(f"SSL cert file not found: {config.ssl.cert_file}")
|
|
if not Path(config.ssl.key_file).exists():
|
|
warnings.append(f"SSL key file not found: {config.ssl.key_file}")
|
|
|
|
valid_extension_types = [
|
|
"routing",
|
|
"process_orchestration",
|
|
"asgi_mount",
|
|
]
|
|
for ext in config.extensions:
|
|
if ext.type not in valid_extension_types:
|
|
warnings.append(f"Unknown extension type: {ext.type}")
|
|
|
|
if strict:
|
|
known_top_level = {"http", "server", "ssl", "logging", "extensions"}
|
|
for key in data.keys():
|
|
if key not in known_top_level:
|
|
warnings.append(f"Unknown top-level field: {key}")
|
|
|
|
if errors:
|
|
for error in errors:
|
|
print_error(error)
|
|
raise click.Abort()
|
|
|
|
if warnings:
|
|
for warning in warnings:
|
|
print_warning(warning)
|
|
|
|
print_success("Configuration is valid!")
|
|
|
|
except yaml.YAMLError as e:
|
|
print_error(f"YAML syntax error: {e}")
|
|
raise click.Abort()
|
|
except Exception as e:
|
|
print_error(f"Validation error: {e}")
|
|
raise click.Abort()
|
|
|
|
|
|
@config_cmd.command("show")
|
|
@click.option(
|
|
"-c",
|
|
"--config",
|
|
"config_file",
|
|
default=None,
|
|
help="Path to configuration file",
|
|
)
|
|
@click.option(
|
|
"--format",
|
|
"output_format",
|
|
type=click.Choice(["yaml", "json", "table"]),
|
|
default="yaml",
|
|
help="Output format",
|
|
)
|
|
@click.option(
|
|
"--section",
|
|
"section",
|
|
default=None,
|
|
help="Show only a specific section (e.g., server, logging)",
|
|
)
|
|
@click.pass_obj
|
|
def show_cmd(ctx, config_file: Optional[str], output_format: str, section: Optional[str]):
|
|
"""
|
|
Display current configuration.
|
|
|
|
\b
|
|
Examples:
|
|
pyserve config show
|
|
pyserve config show --format json
|
|
pyserve config show --section server
|
|
"""
|
|
from ..output import console, print_error
|
|
|
|
config_path = Path(config_file or ctx.config_file)
|
|
|
|
if not config_path.exists():
|
|
print_error(f"Configuration file not found: {config_path}")
|
|
raise click.Abort()
|
|
|
|
try:
|
|
with open(config_path) as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
if section:
|
|
if section in data:
|
|
data = {section: data[section]}
|
|
else:
|
|
print_error(f"Section '{section}' not found in configuration")
|
|
raise click.Abort()
|
|
|
|
if output_format == "yaml":
|
|
from rich.syntax import Syntax
|
|
|
|
yaml_str = yaml.dump(data, default_flow_style=False, sort_keys=False)
|
|
syntax = Syntax(yaml_str, "yaml", theme="monokai", line_numbers=False)
|
|
console.print(syntax)
|
|
|
|
elif output_format == "json":
|
|
from rich.syntax import Syntax
|
|
|
|
json_str = json.dumps(data, indent=2)
|
|
syntax = Syntax(json_str, "json", theme="monokai", line_numbers=False)
|
|
console.print(syntax)
|
|
|
|
elif output_format == "table":
|
|
from rich.tree import Tree
|
|
|
|
def build_tree(data, tree):
|
|
if isinstance(data, dict):
|
|
for key, value in data.items():
|
|
if isinstance(value, (dict, list)):
|
|
branch = tree.add(f"[cyan]{key}[/cyan]")
|
|
build_tree(value, branch)
|
|
else:
|
|
tree.add(f"[cyan]{key}[/cyan]: [green]{value}[/green]")
|
|
elif isinstance(data, list):
|
|
for i, item in enumerate(data):
|
|
if isinstance(item, (dict, list)):
|
|
branch = tree.add(f"[dim][{i}][/dim]")
|
|
build_tree(item, branch)
|
|
else:
|
|
tree.add(f"[dim][{i}][/dim] [green]{item}[/green]")
|
|
|
|
tree = Tree(f"[bold]Configuration: {config_path}[/bold]")
|
|
build_tree(data, tree)
|
|
console.print(tree)
|
|
|
|
except Exception as e:
|
|
print_error(f"Error reading configuration: {e}")
|
|
raise click.Abort()
|
|
|
|
|
|
@config_cmd.command("get")
|
|
@click.argument("key")
|
|
@click.option(
|
|
"-c",
|
|
"--config",
|
|
"config_file",
|
|
default=None,
|
|
help="Path to configuration file",
|
|
)
|
|
@click.pass_obj
|
|
def get_cmd(ctx, key: str, config_file: Optional[str]):
|
|
"""
|
|
Get a specific configuration value.
|
|
|
|
Use dot notation to access nested values.
|
|
|
|
\b
|
|
Examples:
|
|
pyserve config get server.port
|
|
pyserve config get logging.level
|
|
pyserve config get extensions.0.type
|
|
"""
|
|
from ..output import console, print_error
|
|
|
|
config_path = Path(config_file or ctx.config_file)
|
|
|
|
if not config_path.exists():
|
|
print_error(f"Configuration file not found: {config_path}")
|
|
raise click.Abort()
|
|
|
|
try:
|
|
with open(config_path) as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
value = data
|
|
for part in key.split("."):
|
|
if isinstance(value, dict):
|
|
if part in value:
|
|
value = value[part]
|
|
else:
|
|
print_error(f"Key '{key}' not found")
|
|
raise click.Abort()
|
|
elif isinstance(value, list):
|
|
try:
|
|
index = int(part)
|
|
value = value[index]
|
|
except (ValueError, IndexError):
|
|
print_error(f"Invalid index '{part}' in key '{key}'")
|
|
raise click.Abort()
|
|
else:
|
|
print_error(f"Cannot access '{part}' in {type(value).__name__}")
|
|
raise click.Abort()
|
|
|
|
if isinstance(value, (dict, list)):
|
|
console.print(yaml.dump(value, default_flow_style=False))
|
|
else:
|
|
console.print(str(value))
|
|
|
|
except Exception as e:
|
|
print_error(f"Error: {e}")
|
|
raise click.Abort()
|
|
|
|
|
|
@config_cmd.command("set")
|
|
@click.argument("key")
|
|
@click.argument("value")
|
|
@click.option(
|
|
"-c",
|
|
"--config",
|
|
"config_file",
|
|
default=None,
|
|
help="Path to configuration file",
|
|
)
|
|
@click.pass_obj
|
|
def set_cmd(ctx, key: str, value: str, config_file: Optional[str]):
|
|
"""
|
|
Set a configuration value.
|
|
|
|
Use dot notation to access nested values.
|
|
|
|
\b
|
|
Examples:
|
|
pyserve config set server.port 8080
|
|
pyserve config set logging.level DEBUG
|
|
"""
|
|
from ..output import print_error, print_success
|
|
|
|
config_path = Path(config_file or ctx.config_file)
|
|
|
|
if not config_path.exists():
|
|
print_error(f"Configuration file not found: {config_path}")
|
|
raise click.Abort()
|
|
|
|
try:
|
|
with open(config_path) as f:
|
|
data = yaml.safe_load(f)
|
|
|
|
parsed_value: Any
|
|
if value.lower() == "true":
|
|
parsed_value = True
|
|
elif value.lower() == "false":
|
|
parsed_value = False
|
|
elif value.isdigit():
|
|
parsed_value = int(value)
|
|
else:
|
|
try:
|
|
parsed_value = float(value)
|
|
except ValueError:
|
|
parsed_value = value
|
|
|
|
parts = key.split(".")
|
|
current = data
|
|
for part in parts[:-1]:
|
|
if isinstance(current, dict):
|
|
if part not in current:
|
|
current[part] = {}
|
|
current = current[part]
|
|
elif isinstance(current, list):
|
|
index = int(part)
|
|
current = current[index]
|
|
|
|
final_key = parts[-1]
|
|
if isinstance(current, dict):
|
|
current[final_key] = parsed_value
|
|
elif isinstance(current, list):
|
|
current[int(final_key)] = parsed_value
|
|
|
|
with open(config_path, "w") as f:
|
|
yaml.dump(data, f, default_flow_style=False, sort_keys=False)
|
|
|
|
print_success(f"Set {key} = {parsed_value}")
|
|
|
|
except Exception as e:
|
|
print_error(f"Error: {e}")
|
|
raise click.Abort()
|
|
|
|
|
|
@config_cmd.command("diff")
|
|
@click.argument("file1", type=click.Path(exists=True))
|
|
@click.argument("file2", type=click.Path(exists=True))
|
|
def diff_cmd(file1: str, file2: str):
|
|
"""
|
|
Compare two configuration files.
|
|
|
|
\b
|
|
Examples:
|
|
pyserve config diff config.yaml production.yaml
|
|
"""
|
|
from ..output import console, print_error
|
|
|
|
try:
|
|
with open(file1) as f:
|
|
data1 = yaml.safe_load(f)
|
|
with open(file2) as f:
|
|
data2 = yaml.safe_load(f)
|
|
|
|
def compare_dicts(d1, d2, path=""):
|
|
differences = []
|
|
|
|
all_keys = set(d1.keys() if d1 else []) | set(d2.keys() if d2 else [])
|
|
|
|
for key in sorted(all_keys):
|
|
current_path = f"{path}.{key}" if path else key
|
|
v1 = d1.get(key) if d1 else None
|
|
v2 = d2.get(key) if d2 else None
|
|
|
|
if key not in (d1 or {}):
|
|
differences.append(("added", current_path, None, v2))
|
|
elif key not in (d2 or {}):
|
|
differences.append(("removed", current_path, v1, None))
|
|
elif isinstance(v1, dict) and isinstance(v2, dict):
|
|
differences.extend(compare_dicts(v1, v2, current_path))
|
|
elif v1 != v2:
|
|
differences.append(("changed", current_path, v1, v2))
|
|
|
|
return differences
|
|
|
|
differences = compare_dicts(data1, data2)
|
|
|
|
if not differences:
|
|
console.print("[green]Files are identical[/green]")
|
|
return
|
|
|
|
console.print(f"\n[bold]Differences between {file1} and {file2}:[/bold]\n")
|
|
|
|
for diff_type, path, v1, v2 in differences:
|
|
if diff_type == "added":
|
|
console.print(f" [green]+ {path}: {v2}[/green]")
|
|
elif diff_type == "removed":
|
|
console.print(f" [red]- {path}: {v1}[/red]")
|
|
elif diff_type == "changed":
|
|
console.print(f" [yellow]~ {path}:[/yellow]")
|
|
console.print(f" [red]- {v1}[/red]")
|
|
console.print(f" [green]+ {v2}[/green]")
|
|
|
|
console.print()
|
|
|
|
except Exception as e:
|
|
print_error(f"Error: {e}")
|
|
raise click.Abort()
|