163 lines
5.2 KiB
Python
163 lines
5.2 KiB
Python
"""Core validation engine for environment variables."""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
from envschema.schema import Schema
|
|
from envschema.loader import EnvLoader
|
|
from envschema.validators import validate_value
|
|
|
|
|
|
@dataclass
|
|
class ValidationError:
|
|
"""Represents a validation error for a specific variable."""
|
|
|
|
var_name: str
|
|
error_type: str
|
|
message: str
|
|
value: Optional[str] = None
|
|
|
|
def to_dict(self) -> dict:
|
|
"""Convert to dictionary for JSON output."""
|
|
return {
|
|
"var_name": self.var_name,
|
|
"error_type": self.error_type,
|
|
"message": self.message,
|
|
"value": self.value,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class ValidationResult:
|
|
"""Result of schema validation."""
|
|
|
|
is_valid: bool
|
|
missing_required: list[str] = field(default_factory=list)
|
|
type_errors: list[ValidationError] = field(default_factory=list)
|
|
pattern_errors: list[ValidationError] = field(default_factory=list)
|
|
warnings: list[str] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict:
|
|
"""Convert to dictionary for JSON output."""
|
|
return {
|
|
"is_valid": self.is_valid,
|
|
"missing_required": self.missing_required,
|
|
"type_errors": [e.to_dict() for e in self.type_errors],
|
|
"pattern_errors": [e.to_dict() for e in self.pattern_errors],
|
|
"warnings": self.warnings,
|
|
}
|
|
|
|
|
|
class ValidationEngine:
|
|
"""Engine for validating environment variables against a schema."""
|
|
|
|
def __init__(self, schema: Schema):
|
|
"""Initialize the validation engine.
|
|
|
|
Args:
|
|
schema: The schema to validate against.
|
|
"""
|
|
self.schema = schema
|
|
|
|
def validate(self, env_vars: dict[str, str]) -> ValidationResult:
|
|
"""Validate environment variables against the schema.
|
|
|
|
Args:
|
|
env_vars: Dictionary of environment variable names to values.
|
|
|
|
Returns:
|
|
ValidationResult with all errors and warnings.
|
|
"""
|
|
result = ValidationResult(is_valid=True)
|
|
|
|
self._check_required_vars(env_vars, result)
|
|
self._validate_types(env_vars, result)
|
|
self._check_extra_vars(env_vars, result)
|
|
|
|
result.is_valid = (
|
|
len(result.missing_required) == 0
|
|
and len(result.type_errors) == 0
|
|
and len(result.pattern_errors) == 0
|
|
)
|
|
|
|
return result
|
|
|
|
def _check_required_vars(self, env_vars: dict[str, str], result: ValidationResult) -> None:
|
|
"""Check for missing required variables.
|
|
|
|
Args:
|
|
env_vars: Environment variables.
|
|
result: Validation result to update.
|
|
"""
|
|
required_vars = self.schema.get_required_vars()
|
|
env_keys_upper = {k.upper() for k in env_vars.keys()}
|
|
|
|
for var in required_vars:
|
|
if var.name.upper() not in env_keys_upper:
|
|
result.missing_required.append(var.name)
|
|
result.is_valid = False
|
|
|
|
def _validate_types(self, env_vars: dict[str, str], result: ValidationResult) -> None:
|
|
"""Validate types of environment variables.
|
|
|
|
Args:
|
|
env_vars: Environment variables.
|
|
result: Validation result to update.
|
|
"""
|
|
env_vars_upper = {k.upper(): v for k, v in env_vars.items()}
|
|
for var in self.schema.envvars:
|
|
value = env_vars_upper.get(var.name.upper())
|
|
|
|
if value is None and var.default is not None:
|
|
continue
|
|
|
|
if value is not None:
|
|
is_valid, error = validate_value(value, var.type, var.pattern)
|
|
if not is_valid and error:
|
|
result.type_errors.append(
|
|
ValidationError(
|
|
var_name=var.name,
|
|
error_type="type_mismatch",
|
|
message=error.message,
|
|
value=error.value,
|
|
)
|
|
)
|
|
result.is_valid = False
|
|
|
|
def _check_extra_vars(self, env_vars: dict[str, str], result: ValidationResult) -> None:
|
|
"""Check for extra variables not in schema (warning only).
|
|
|
|
Args:
|
|
env_vars: Environment variables.
|
|
result: Validation result to update.
|
|
"""
|
|
schema_keys_upper = {v.name.upper() for v in self.schema.envvars}
|
|
for key in env_vars.keys():
|
|
if key.upper() not in schema_keys_upper:
|
|
result.warnings.append(f"Unknown environment variable: {key}")
|
|
|
|
|
|
def validate_environment(
|
|
schema_path: str,
|
|
env_file: Optional[str] = None,
|
|
use_environment: bool = True,
|
|
) -> ValidationResult:
|
|
"""Convenience function to validate environment against a schema file.
|
|
|
|
Args:
|
|
schema_path: Path to the schema file (JSON or YAML).
|
|
env_file: Optional path to .env file.
|
|
use_environment: Whether to include os.environ.
|
|
|
|
Returns:
|
|
ValidationResult with validation status.
|
|
"""
|
|
from envschema.schema import load_schema_from_file
|
|
|
|
schema = load_schema_from_file(schema_path)
|
|
loader = EnvLoader(file_path=env_file, use_environment=use_environment)
|
|
env_vars = loader.load()
|
|
|
|
engine = ValidationEngine(schema)
|
|
return engine.validate(env_vars)
|