Initial upload with full project structure
This commit is contained in:
149
app/src/confgen/parsers.py
Normal file
149
app/src/confgen/parsers.py
Normal file
@@ -0,0 +1,149 @@
|
||||
"""Configuration parsers for JSON, YAML, and TOML formats."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
import sys
|
||||
|
||||
|
||||
class ConfigParser:
|
||||
"""Parser for different configuration formats."""
|
||||
|
||||
SUPPORTED_FORMATS = ("json", "yaml", "toml")
|
||||
|
||||
def __init__(self):
|
||||
if sys.version_info >= (3, 11):
|
||||
import tomllib
|
||||
|
||||
self.toml_module = tomllib
|
||||
self.toml_loads = tomllib.loads
|
||||
else:
|
||||
self.toml_module = None
|
||||
self.toml_loads = None
|
||||
|
||||
def detect_format(self, content: str) -> str:
|
||||
"""Detect the format of configuration content."""
|
||||
content = content.strip()
|
||||
|
||||
if content.startswith("{") or content.startswith("["):
|
||||
try:
|
||||
json.loads(content)
|
||||
return "json"
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
if content.startswith("#") or content.startswith("---"):
|
||||
return "yaml"
|
||||
|
||||
import yaml
|
||||
try:
|
||||
yaml.safe_load(content)
|
||||
return "yaml"
|
||||
except yaml.YAMLError:
|
||||
pass
|
||||
|
||||
if self.toml_loads:
|
||||
try:
|
||||
self.toml_loads(content)
|
||||
return "toml"
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return "yaml"
|
||||
|
||||
def detect_format_from_path(self, path: str) -> str:
|
||||
"""Detect format from file extension."""
|
||||
ext = Path(path).suffix.lower()
|
||||
format_map = {
|
||||
".json": "json",
|
||||
".yaml": "yaml",
|
||||
".yml": "yaml",
|
||||
".toml": "toml",
|
||||
}
|
||||
return format_map.get(ext, "yaml")
|
||||
|
||||
def parse(self, content: str, format: str | None = None) -> dict[str, Any]:
|
||||
"""Parse configuration content."""
|
||||
fmt = format or self.detect_format(content)
|
||||
|
||||
if fmt == "json":
|
||||
return self.parse_json(content)
|
||||
elif fmt == "toml":
|
||||
return self.parse_toml(content)
|
||||
else:
|
||||
return self.parse_yaml(content)
|
||||
|
||||
def parse_json(self, content: str) -> dict[str, Any]:
|
||||
"""Parse JSON configuration."""
|
||||
try:
|
||||
return json.loads(content)
|
||||
except json.JSONDecodeError as e:
|
||||
raise ValueError(f"Invalid JSON: {e}")
|
||||
|
||||
def parse_yaml(self, content: str) -> dict[str, Any]:
|
||||
"""Parse YAML configuration."""
|
||||
import yaml
|
||||
try:
|
||||
return yaml.safe_load(content) or {}
|
||||
except yaml.YAMLError as e:
|
||||
raise ValueError(f"Invalid YAML: {e}")
|
||||
|
||||
def parse_toml(self, content: str) -> dict[str, Any]:
|
||||
"""Parse TOML configuration."""
|
||||
if not self.toml_loads:
|
||||
raise ValueError("TOML parsing not available. Install 'tomli' for Python < 3.11")
|
||||
try:
|
||||
return self.toml_loads(content)
|
||||
except Exception as e:
|
||||
raise ValueError(f"Invalid TOML: {e}")
|
||||
|
||||
def to_json(self, data: dict[str, Any], indent: int = 2) -> str:
|
||||
"""Convert data to JSON format."""
|
||||
return json.dumps(data, indent=indent, ensure_ascii=False)
|
||||
|
||||
def to_yaml(self, data: dict[str, Any], indent: int = 2) -> str:
|
||||
"""Convert data to YAML format."""
|
||||
import yaml
|
||||
return yaml.dump(data, indent=indent, allow_unicode=True, sort_keys=False)
|
||||
|
||||
def to_toml(self, data: dict[str, Any]) -> str:
|
||||
"""Convert data to TOML format."""
|
||||
try:
|
||||
import tomli_w
|
||||
return tomli_w.dumps(data)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
try:
|
||||
import tomlkit
|
||||
return tomlkit.dumps(data)
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
raise ValueError("TOML output not available. Install 'tomli-w' or 'tomlkit'")
|
||||
|
||||
def load_file(self, path: str) -> dict[str, Any]:
|
||||
"""Load and parse a configuration file."""
|
||||
fmt = self.detect_format_from_path(path)
|
||||
with open(path) as f:
|
||||
content = f.read()
|
||||
return self.parse(content, fmt)
|
||||
|
||||
def save_file(
|
||||
self,
|
||||
path: str,
|
||||
data: dict[str, Any],
|
||||
format: str | None = None,
|
||||
) -> None:
|
||||
"""Save data to a configuration file."""
|
||||
fmt = format or self.detect_format_from_path(path)
|
||||
|
||||
if fmt == "json":
|
||||
content = self.to_json(data)
|
||||
elif fmt == "toml":
|
||||
content = self.to_toml(data)
|
||||
else:
|
||||
content = self.to_yaml(data)
|
||||
|
||||
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
||||
Path(path).write_text(content)
|
||||
Reference in New Issue
Block a user