forked from aegis/pyserveX
pyservectl init
This commit is contained in:
Илья Глазунов
@@ -0,0 +1,420 @@
|
||||
"""
|
||||
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.table import 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()
|
||||
Reference in New Issue
Block a user