Initial upload: ConfDoc v0.1.0 - Config validation and documentation generator
This commit is contained in:
179
src/confdoc/profiles/manager.py
Normal file
179
src/confdoc/profiles/manager.py
Normal file
@@ -0,0 +1,179 @@
|
||||
import os
|
||||
from typing import Any, Dict, List, Optional
|
||||
import json
|
||||
|
||||
|
||||
class ProfileManager:
|
||||
"""Manages environment-specific configuration profiles."""
|
||||
|
||||
def __init__(self, profile_dir: Optional[str] = None):
|
||||
"""Initialize profile manager with optional profile directory."""
|
||||
self.profile_dir = profile_dir or os.path.expanduser("~/.config/confdoc/profiles")
|
||||
self._profiles: Dict[str, Dict[str, Any]] = {}
|
||||
self._load_builtin_profiles()
|
||||
|
||||
def _load_builtin_profiles(self):
|
||||
"""Load built-in profiles."""
|
||||
self._profiles = {
|
||||
"development": {
|
||||
"name": "Development",
|
||||
"description": "Development environment settings",
|
||||
"validation": {
|
||||
"strict": False,
|
||||
"allow_extra": True,
|
||||
},
|
||||
"overrides": {
|
||||
"debug": True,
|
||||
"log_level": "DEBUG",
|
||||
}
|
||||
},
|
||||
"staging": {
|
||||
"name": "Staging",
|
||||
"description": "Staging environment settings",
|
||||
"validation": {
|
||||
"strict": True,
|
||||
"allow_extra": False,
|
||||
},
|
||||
"overrides": {
|
||||
"debug": False,
|
||||
"log_level": "INFO",
|
||||
}
|
||||
},
|
||||
"production": {
|
||||
"name": "Production",
|
||||
"description": "Production environment settings",
|
||||
"validation": {
|
||||
"strict": True,
|
||||
"allow_extra": False,
|
||||
},
|
||||
"overrides": {
|
||||
"debug": False,
|
||||
"log_level": "WARNING",
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
def list_profiles(self) -> List[str]:
|
||||
"""List available profile names."""
|
||||
return list(self._profiles.keys())
|
||||
|
||||
def get_profile_info(self, name: str) -> Optional[Dict[str, Any]]:
|
||||
"""Get profile information."""
|
||||
return self._profiles.get(name)
|
||||
|
||||
def load_profile(self, name: str) -> Optional[Dict[str, Any]]:
|
||||
"""Load a profile by name."""
|
||||
return self._profiles.get(name)
|
||||
|
||||
def save_profile(self, name: str, profile: Dict[str, Any]) -> None:
|
||||
"""Save a custom profile."""
|
||||
self._profiles[name] = profile
|
||||
|
||||
if self.profile_dir:
|
||||
os.makedirs(self.profile_dir, exist_ok=True)
|
||||
profile_path = os.path.join(self.profile_dir, f"{name}.json")
|
||||
with open(profile_path, 'w') as f:
|
||||
json.dump(profile, f, indent=2)
|
||||
|
||||
def apply_profile(self, schema: Dict[str, Any], profile: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Apply profile modifications to a schema."""
|
||||
if "validation" in profile:
|
||||
schema = self._apply_validation_rules(schema, profile["validation"])
|
||||
|
||||
if "overrides" in profile:
|
||||
schema = self._apply_overrides(schema, profile["overrides"])
|
||||
|
||||
return schema
|
||||
|
||||
def _apply_validation_rules(self, schema: Dict[str, Any], rules: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Apply validation rules from profile to schema."""
|
||||
schema = schema.copy()
|
||||
|
||||
if "strict" in rules:
|
||||
if rules["strict"]:
|
||||
schema.setdefault("additionalProperties", False)
|
||||
else:
|
||||
schema.pop("additionalProperties", None)
|
||||
|
||||
if "allow_extra" in rules and rules["allow_extra"]:
|
||||
schema.pop("additionalProperties", None)
|
||||
|
||||
return schema
|
||||
|
||||
def _apply_overrides(self, schema: Dict[str, Any], overrides: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Apply default value overrides from profile."""
|
||||
schema = schema.copy()
|
||||
|
||||
if "properties" not in schema:
|
||||
schema["properties"] = {}
|
||||
|
||||
for key, value in overrides.items():
|
||||
if key in schema.get("properties", {}):
|
||||
if "default" not in schema["properties"][key]:
|
||||
schema["properties"][key]["default"] = value
|
||||
else:
|
||||
schema["properties"][key] = {
|
||||
"type": type(value).__name__,
|
||||
"default": value,
|
||||
"description": f"Default value for {key}",
|
||||
}
|
||||
|
||||
return schema
|
||||
|
||||
def create_profile_from_config(self, config: Dict[str, Any], name: str, description: str = "") -> Dict[str, Any]:
|
||||
"""Create a profile from an existing configuration."""
|
||||
profile = {
|
||||
"name": name,
|
||||
"description": description,
|
||||
"derived_from": config,
|
||||
"validation": {
|
||||
"strict": True,
|
||||
"allow_extra": False,
|
||||
},
|
||||
"overrides": {},
|
||||
}
|
||||
|
||||
for key, value in config.items():
|
||||
if isinstance(value, (str, int, bool)):
|
||||
profile["overrides"][key] = value
|
||||
|
||||
return profile
|
||||
|
||||
def merge_profiles(self, base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Merge two profiles, with override taking precedence."""
|
||||
result = base.copy()
|
||||
|
||||
for key, value in override.items():
|
||||
if key in ("validation", "overrides") and key in result and isinstance(value, dict):
|
||||
result[key] = {**result[key], **value}
|
||||
else:
|
||||
result[key] = value
|
||||
|
||||
return result
|
||||
|
||||
def get_environment_overrides(self, env_vars: Dict[str, str]) -> Dict[str, Any]:
|
||||
"""Extract configuration overrides from environment variables."""
|
||||
overrides = {}
|
||||
|
||||
for key, value in env_vars.items():
|
||||
if key.startswith("CONFDOC_"):
|
||||
config_key = key[8:].lower()
|
||||
parsed_value = self._parse_env_value(value)
|
||||
overrides[config_key] = parsed_value
|
||||
|
||||
return overrides
|
||||
|
||||
def _parse_env_value(self, value: str) -> Any:
|
||||
"""Parse environment variable value to appropriate type."""
|
||||
value = value.strip()
|
||||
|
||||
if value.lower() == "true":
|
||||
return True
|
||||
elif value.lower() == "false":
|
||||
return False
|
||||
elif value.isdigit():
|
||||
return int(value)
|
||||
elif value.replace(".", "", 1).isdigit():
|
||||
return float(value)
|
||||
|
||||
return value
|
||||
Reference in New Issue
Block a user