envschema/envschema/core.py

"""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)