From 82516ee4c8dbdd44e3b06aeb4c6c0d454537cf65 Mon Sep 17 00:00:00 2001 From: 7000pctAUTO Date: Sun, 22 Mar 2026 15:13:53 +0000 Subject: [PATCH] Initial upload: EnvSchema v0.1.0 with CI/CD workflow --- envschema/core.py | 162 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 162 insertions(+) create mode 100644 envschema/core.py diff --git a/envschema/core.py b/envschema/core.py new file mode 100644 index 0000000..c841567 --- /dev/null +++ b/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) \ No newline at end of file