Add env_pro core modules

app/env_pro/core/validator.py

@@ -0,0 +1,116 @@
+"""Schema validation for environment variables."""
+
+import re
+from pathlib import Path
+from typing import Optional, List, Dict, Any
+from pydantic import BaseModel, Field, validator
+from pydantic.types import AnyUrl
+
+
+class VarSchema(BaseModel):
+    """Schema for a single variable."""
+    type: str = "string"
+    required: bool = False
+    description: Optional[str] = None
+    default: Optional[str] = None
+    pattern: Optional[str] = None
+    min_length: Optional[int] = None
+    max_length: Optional[int] = None
+
+
+class Schema(BaseModel):
+    """Full schema for environment validation."""
+    variables: Dict[str, VarSchema] = {}
+
+    class Config:
+        arbitrary_types_allowed = True
+
+
+class ValidationError:
+    """Represents a validation error."""
+
+    def __init__(self, key: str, message: str, severity: str = "error"):
+        self.key = key
+        self.message = message
+        self.severity = severity
+
+    def __str__(self):
+        return f"{self.key}: {self.message}"
+
+
+class Validator:
+    """Environment variable validator."""
+
+    def __init__(self, schema: Optional[Schema] = None):
+        """Initialize validator with optional schema."""
+        self.schema = schema
+
+    @classmethod
+    def from_file(cls, schema_path: Optional[Path] = None) -> "Validator":
+        """Load schema from file."""
+        import yaml
+
+        if schema_path is None:
+            schema_path = Path.cwd() / ".env.schema.yaml"
+
+        if not schema_path.exists():
+            return cls()
+
+        try:
+            with open(schema_path, 'r') as f:
+                data = yaml.safe_load(f)
+            return cls(schema=Schema(**data))
+        except yaml.YAMLError:
+            return cls()
+
+    def validate(self, variables: Dict[str, str]) -> List[ValidationError]:
+        """Validate variables against schema."""
+        errors = []
+
+        if self.schema is None:
+            return errors
+
+        for key, var_schema in self.schema.variables.items():
+            value = variables.get(key)
+
+            if var_schema.required and (value is None or value == ""):
+                errors.append(ValidationError(key, "required variable is missing"))
+                continue
+
+            if value is not None:
+                if var_schema.pattern:
+                    if not re.match(var_schema.pattern, value):
+                        errors.append(ValidationError(key, f"value does not match pattern {var_schema.pattern}"))
+
+                if var_schema.min_length and len(value) < var_schema.min_length:
+                    errors.append(ValidationError(key, f"value length {len(value)} is less than minimum {var_schema.min_length}"))
+
+                if var_schema.max_length and len(value) > var_schema.max_length:
+                    errors.append(ValidationError(key, f"value length {len(value)} exceeds maximum {var_schema.max_length}"))
+
+        return errors
+
+    def check_types(self, variables: Dict[str, str]) -> List[ValidationError]:
+        """Check variable types against schema."""
+        errors = []
+
+        if self.schema is None:
+            return errors
+
+        type_converters = {
+            "bool": lambda v: v.lower() in ("true", "1", "yes"),
+            "int": lambda v: int(v),
+            "float": lambda v: float(v),
+            "url": lambda v: AnyUrl(v),
+        }
+
+        for key, var_schema in self.schema.variables.items():
+            value = variables.get(key)
+
+            if value is not None and var_schema.type in type_converters:
+                try:
+                    type_converters[var_schema.type](value)
+                except (ValueError, Exception):
+                    errors.append(ValidationError(key, f"cannot convert to {var_schema.type}"))
+
+        return errors