Initial upload: EnvSchema v0.1.0 with CI/CD workflow

envschema/core.py

@@ -0,0 +1,162 @@
+"""Core validation engine for environment variables."""
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+from envschema.schema import Schema, EnvVar
+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)