schema-validation

Schema Validation for Construction Data

Overview

Validate data structures against defined schemas for construction data exchange. Ensure API payloads, file imports, and BIM exports conform to expected formats before processing.

Schema Validation Framework

Core Schema Validator

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union
from enum import Enum
import json
import re
from datetime import datetime

class SchemaType(Enum):
    STRING = "string"
    NUMBER = "number"
    INTEGER = "integer"
    BOOLEAN = "boolean"
    ARRAY = "array"
    OBJECT = "object"
    DATE = "date"
    DATETIME = "datetime"
    CSI_CODE = "csi_code"
    CURRENCY = "currency"
    GUID = "guid"

@dataclass
class SchemaField:
    name: str
    type: SchemaType
    required: bool = True
    nullable: bool = False
    min_value: Optional[float] = None
    max_value: Optional[float] = None
    min_length: Optional[int] = None
    max_length: Optional[int] = None
    pattern: Optional[str] = None
    enum_values: Optional[List[Any]] = None
    items_schema: Optional['Schema'] = None  # For arrays
    properties: Optional[Dict[str, 'SchemaField']] = None  # For objects
    description: str = ""

@dataclass
class Schema:
    name: str
    version: str
    fields: Dict[str, SchemaField]
    description: str = ""

@dataclass
class SchemaValidationError:
    path: str
    message: str
    expected: str
    actual: Any

@dataclass
class SchemaValidationResult:
    is_valid: bool
    errors: List[SchemaValidationError] = field(default_factory=list)
    schema_name: str = ""
    schema_version: str = ""

    def add_error(self, path: str, message: str, expected: str, actual: Any):
        self.errors.append(SchemaValidationError(path, message, expected, actual))
        self.is_valid = False

    def to_report(self) -> str:
        lines = [
            f"Schema Validation: {self.schema_name} v{self.schema_version}",
            "=" * 50,
            f"Status: {'✓ VALID' if self.is_valid else '✗ INVALID'}",
            f"Errors: {len(self.errors)}",
            ""
        ]

        for error in self.errors:
            lines.append(f"❌ {error.path}")
            lines.append(f"   {error.message}")
            lines.append(f"   Expected: {error.expected}")
            lines.append(f"   Actual: {error.actual}")
            lines.append("")

        return "\n".join(lines)


class SchemaValidator:
    """Validate data against schemas."""

    # Custom type patterns
    PATTERNS = {
        SchemaType.CSI_CODE: r'^\d{2}\s?\d{2}\s?\d{2}$',
        SchemaType.GUID: r'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$',
        SchemaType.CURRENCY: r'^-?\d+(\.\d{2})?$',
        SchemaType.DATE: r'^\d{4}-\d{2}-\d{2}$',
        SchemaType.DATETIME: r'^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}',
    }

    def validate(self, data: Any, schema: Schema) -> SchemaValidationResult:
        result = SchemaValidationResult(
            is_valid=True,
            schema_name=schema.name,
            schema_version=schema.version
        )

        self._validate_object(data, schema.fields, "", result)
        return result

    def _validate_object(self, data: Any, fields: Dict[str, SchemaField], path: str, result: SchemaValidationResult):
        if not isinstance(data, dict):
            result.add_error(path or "root", "Expected object", "object", type(data).__name__)
            return

        # Check required fields
        for field_name, field_schema in fields.items():
            field_path = f"{path}.{field_name}" if path else field_name

            if field_name not in data:
                if field_schema.required:
                    result.add_error(field_path, "Required field missing", "present", "missing")
                continue

            value = data[field_name]

            # Check nullable
            if value is None:
                if not field_schema.nullable:
                    result.add_error(field_path, "Field cannot be null", "non-null", "null")
                continue

            # Validate type
            self._validate_field(value, field_schema, field_path, result)

        # Check for extra fields (warning only)
        for key in data.keys():
            if key not in fields:
                # Could add warning here if needed
                pass

    def _validate_field(self, value: Any, schema: SchemaField, path: str, result: SchemaValidationResult):
        # Type validation
        if not self._check_type(value, schema.type):
            result.add_error(path, f"Invalid type", schema.type.value, type(value).__name__)
            return

        # String validations
        if schema.type == SchemaType.STRING:
            if schema.min_length and len(value) < schema.min_length:
                result.add_error(path, f"String too short", f"min {schema.min_length}", len(value))
            if schema.max_length and len(value) > schema.max_length:
                result.add_error(path, f"String too long", f"max {schema.max_length}", len(value))
            if schema.pattern and not re.match(schema.pattern, value):
                result.add_error(path, "Pattern mismatch", schema.pattern, value)

        # Numeric validations
        if schema.type in (SchemaType.NUMBER, SchemaType.INTEGER):
            if schema.min_value is not None and value < schema.min_value:
                result.add_error(path, "Value below minimum", f">= {schema.min_value}", value)
            if schema.max_value is not None and value > schema.max_value:
                result.add_error(path, "Value above maximum", f"<= {schema.max_value}", value)

        # Enum validation
        if schema.enum_values and value not in schema.enum_values:
            result.add_error(path, "Invalid enum value", str(schema.enum_values), value)

        # Array validation
        if schema.type == SchemaType.ARRAY and schema.items_schema:
            for i, item in enumerate(value):
                item_path = f"{path}[{i}]"
                if schema.items_schema.fields:
                    self._validate_object(item, schema.items_schema.fields, item_path, result)

        # Nested object validation
        if schema.type == SchemaType.OBJECT and schema.properties:
            self._validate_object(value, schema.properties, path, result)

        # Custom type validation
        if schema.type in self.PATTERNS:
            pattern = self.PATTERNS[schema.type]
            if not re.match(pattern, str(value)):
                result.add_error(path, f"Invalid {schema.type.value} format", pattern, value)

    def _check_type(self, value: Any, expected: SchemaType) -> bool:
        type_checks = {
            SchemaType.STRING: lambda v: isinstance(v, str),
            SchemaType.NUMBER: lambda v: isinstance(v, (int, float)),
            SchemaType.INTEGER: lambda v: isinstance(v, int) and not isinstance(v, bool),
            SchemaType.BOOLEAN: lambda v: isinstance(v, bool),
            SchemaType.ARRAY: lambda v: isinstance(v, list),
            SchemaType.OBJECT: lambda v: isinstance(v, dict),
            SchemaType.DATE: lambda v: isinstance(v, str),
            SchemaType.DATETIME: lambda v: isinstance(v, str),
            SchemaType.CSI_CODE: lambda v: isinstance(v, str),
            SchemaType.CURRENCY: lambda v: isinstance(v, (int, float, str)),
            SchemaType.GUID: lambda v: isinstance(v, str),
        }
        return type_checks.get(expected, lambda v: True)(value)