111 lines
3.4 KiB
Python
111 lines
3.4 KiB
Python
"""Schema models for environment variable definitions."""
|
|
|
|
import json
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import yaml
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
|
|
class EnvVarType(str, Enum):
|
|
"""Supported environment variable types."""
|
|
|
|
STRING = "str"
|
|
INTEGER = "int"
|
|
BOOLEAN = "bool"
|
|
LIST = "list"
|
|
|
|
|
|
class EnvVar(BaseModel):
|
|
"""Definition of a single environment variable."""
|
|
|
|
name: str = Field(..., description="Variable name (e.g., DATABASE_URL)")
|
|
type: EnvVarType = Field(default=EnvVarType.STRING, description="Variable type")
|
|
required: bool = Field(default=False, description="Whether variable is required")
|
|
default: Optional[str] = Field(default=None, description="Default value if optional")
|
|
description: Optional[str] = Field(default=None, description="Variable description")
|
|
pattern: Optional[str] = Field(default=None, description="Regex pattern for validation")
|
|
|
|
@field_validator("name")
|
|
@classmethod
|
|
def name_must_be_valid_env_var(cls, v: str) -> str:
|
|
if not v.replace("_", "").replace("-", "").isalnum():
|
|
raise ValueError("Variable name must contain only alphanumeric characters, underscores, and hyphens")
|
|
return v.upper()
|
|
|
|
|
|
class Schema(BaseModel):
|
|
"""Schema containing all environment variable definitions."""
|
|
|
|
version: Optional[str] = Field(default="1.0", description="Schema version")
|
|
envvars: list[EnvVar] = Field(default_factory=list, alias="envVars")
|
|
|
|
model_config = {"populate_by_name": True}
|
|
|
|
def get_var(self, name: str) -> Optional[EnvVar]:
|
|
"""Get an environment variable by name."""
|
|
name_upper = name.upper()
|
|
for var in self.envvars:
|
|
if var.name.upper() == name_upper:
|
|
return var
|
|
return None
|
|
|
|
def get_required_vars(self) -> list[EnvVar]:
|
|
"""Get all required environment variables."""
|
|
return [var for var in self.envvars if var.required]
|
|
|
|
|
|
def load_schema_from_file(file_path: str) -> Schema:
|
|
"""Load schema from a JSON or YAML file.
|
|
|
|
Args:
|
|
file_path: Path to the schema file
|
|
|
|
Returns:
|
|
Parsed Schema object
|
|
|
|
Raises:
|
|
FileNotFoundError: If schema file doesn't exist
|
|
ValueError: If schema format is invalid
|
|
"""
|
|
path = Path(file_path)
|
|
if not path.exists():
|
|
raise FileNotFoundError(f"Schema file not found: {file_path}")
|
|
|
|
content = path.read_text()
|
|
|
|
if path.suffix.lower() in [".yaml", ".yml"]:
|
|
return load_yaml_schema(content)
|
|
elif path.suffix.lower() == ".json":
|
|
return load_json_schema(content)
|
|
else:
|
|
raise ValueError(f"Unsupported schema format: {path.suffix}. Use .json or .yaml")
|
|
|
|
|
|
def load_json_schema(content: str) -> Schema:
|
|
"""Load schema from JSON content."""
|
|
try:
|
|
data = json.loads(content)
|
|
except json.JSONDecodeError as e:
|
|
raise ValueError(f"Invalid JSON schema: {e}")
|
|
|
|
try:
|
|
return Schema.model_validate(data)
|
|
except Exception as e:
|
|
raise ValueError(f"Invalid schema structure: {e}")
|
|
|
|
|
|
def load_yaml_schema(content: str) -> Schema:
|
|
"""Load schema from YAML content."""
|
|
try:
|
|
data = yaml.safe_load(content)
|
|
except yaml.YAMLError as e:
|
|
raise ValueError(f"Invalid YAML schema: {e}")
|
|
|
|
try:
|
|
return Schema.model_validate(data)
|
|
except Exception as e:
|
|
raise ValueError(f"Invalid schema structure: {e}")
|