Initial upload with full project structure
This commit is contained in:
174
app/src/confgen/core.py
Normal file
174
app/src/confgen/core.py
Normal file
@@ -0,0 +1,174 @@
|
||||
"""Core logic for confgen configuration generation."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
@dataclass
|
||||
class TemplateConfig:
|
||||
"""Configuration for a template."""
|
||||
|
||||
name: str
|
||||
path: Path
|
||||
format: Optional[str] = None
|
||||
schema: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class EnvironmentConfig:
|
||||
"""Configuration for an environment."""
|
||||
|
||||
name: str
|
||||
variables: dict
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConfgenConfig:
|
||||
"""Main configuration for confgen."""
|
||||
|
||||
templates: dict[str, TemplateConfig]
|
||||
environments: dict[str, EnvironmentConfig]
|
||||
|
||||
|
||||
@dataclass
|
||||
class GeneratedConfig:
|
||||
"""Result of generating a configuration."""
|
||||
|
||||
content: str
|
||||
format: str
|
||||
schema_validated: bool = False
|
||||
|
||||
|
||||
class ConfgenCore:
|
||||
"""Core class for configuration generation."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
config_dir: Path,
|
||||
templates_dir: Path,
|
||||
output_dir: Path,
|
||||
environment: str = "dev",
|
||||
):
|
||||
self.config_dir = Path(config_dir)
|
||||
self.templates_dir = Path(templates_dir)
|
||||
self.output_dir = Path(output_dir)
|
||||
self.default_environment = environment
|
||||
self.config = self._load_config()
|
||||
|
||||
def _load_config(self) -> ConfgenConfig:
|
||||
"""Load configuration from confgen.yaml."""
|
||||
config_file = self.config_dir / "confgen.yaml"
|
||||
|
||||
if not config_file.exists():
|
||||
return ConfgenConfig(templates={}, environments={})
|
||||
|
||||
with open(config_file) as f:
|
||||
data = yaml.safe_load(f) or {}
|
||||
|
||||
templates = {}
|
||||
for name, tmpl_data in data.get("templates", {}).items():
|
||||
path = Path(tmpl_data.get("path", f"templates/{name}.j2"))
|
||||
if not path.is_absolute():
|
||||
path = self.config_dir / path
|
||||
templates[name] = TemplateConfig(
|
||||
name=name,
|
||||
path=path,
|
||||
format=tmpl_data.get("format"),
|
||||
schema=tmpl_data.get("schema"),
|
||||
)
|
||||
|
||||
environments = {}
|
||||
for name, env_data in data.get("environments", {}).items():
|
||||
environments[name] = EnvironmentConfig(
|
||||
name=name,
|
||||
variables=env_data.get("variables", {}),
|
||||
)
|
||||
|
||||
return ConfgenConfig(templates=templates, environments=environments)
|
||||
|
||||
def get_template(self, name: str) -> Optional[TemplateConfig]:
|
||||
"""Get a template by name."""
|
||||
return self.config.templates.get(name)
|
||||
|
||||
def get_environment(self, name: str) -> Optional[EnvironmentConfig]:
|
||||
"""Get an environment by name."""
|
||||
return self.config.environments.get(name)
|
||||
|
||||
def generate_config(
|
||||
self,
|
||||
template_name: str,
|
||||
environment: str,
|
||||
format: Optional[str] = None,
|
||||
) -> GeneratedConfig:
|
||||
"""Generate a configuration from a template."""
|
||||
from .template import TemplateEngine
|
||||
from .parsers import ConfigParser
|
||||
from .secrets import SecretsResolver
|
||||
from .validator import SchemaValidator
|
||||
|
||||
tmpl = self.get_template(template_name)
|
||||
if not tmpl:
|
||||
raise ValueError(f"Template '{template_name}' not found")
|
||||
|
||||
env_config = self.get_environment(environment)
|
||||
variables = env_config.variables if env_config else {}
|
||||
|
||||
template_path = tmpl.path
|
||||
if not template_path.exists():
|
||||
template_path = self.templates_dir / f"{template_name}.j2"
|
||||
if not template_path.exists():
|
||||
raise ValueError(f"Template file not found: {template_path}")
|
||||
|
||||
template_engine = TemplateEngine()
|
||||
secrets_resolver = SecretsResolver()
|
||||
|
||||
variables["environment"] = environment
|
||||
|
||||
with open(template_path) as f:
|
||||
template_content = f.read()
|
||||
|
||||
content = secrets_resolver.resolve(template_content)
|
||||
rendered_content = template_engine.render(content, variables)
|
||||
|
||||
output_format = format or tmpl.format or "yaml"
|
||||
parser = ConfigParser()
|
||||
|
||||
if output_format == "json":
|
||||
parsed_data = parser.parse_json(rendered_content)
|
||||
output_content = parser.to_json(parsed_data)
|
||||
elif output_format == "toml":
|
||||
parsed_data = parser.parse_toml(rendered_content)
|
||||
output_content = parser.to_toml(parsed_data)
|
||||
else:
|
||||
parsed_data = parser.parse_yaml(rendered_content)
|
||||
output_content = parser.to_yaml(parsed_data)
|
||||
|
||||
schema_validated = False
|
||||
if tmpl.schema:
|
||||
schema_path = tmpl.schema
|
||||
if not Path(schema_path).is_absolute():
|
||||
schema_path = self.config_dir / schema_path
|
||||
|
||||
if schema_path.exists():
|
||||
validator = SchemaValidator(schema_path)
|
||||
is_valid, _ = validator.validate(parsed_data)
|
||||
if not is_valid:
|
||||
raise ValueError("Schema validation failed")
|
||||
schema_validated = True
|
||||
|
||||
return GeneratedConfig(
|
||||
content=output_content,
|
||||
format=output_format,
|
||||
schema_validated=schema_validated,
|
||||
)
|
||||
|
||||
def list_templates(self) -> list[str]:
|
||||
"""List available template names."""
|
||||
return list(self.config.templates.keys())
|
||||
|
||||
def list_environments(self) -> list[str]:
|
||||
"""List available environment names."""
|
||||
return list(self.config.environments.keys())
|
||||
Reference in New Issue
Block a user