CTY Validation

The pyvider.cty.validation module provides utilities for preventing infinite recursion during validation of deeply nested or circular data structures.

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

Key components: - RecursionContext - Tracks visited objects during validation to detect cycles - RecursionDetector - Manages recursion detection state - with_recursion_detection - Decorator that adds recursion detection to validation methods - get_recursion_context() - Returns the current recursion context - clear_recursion_context() - Clears the recursion detection state - validate_config(schema, config) - Convenience function for validating configurations against schemas (raises CtyValidationError on failure)

The recursion detection system is used internally by all types during the validation process. You typically won't need to interact with it directly unless you're implementing custom types that need to participate in cycle detection.

Usage Example:

from pyvider.cty import CtyObject, CtyString
from pyvider.cty.validation import validate_config

schema = CtyObject(attribute_types={"name": CtyString()})
config = {"name": "Alice"}

# This validates and raises on error, but doesn't return the CtyValue
validate_config(schema, config)

# For most use cases, prefer calling validate() directly on the type:
validated_value = schema.validate(config)  # Returns CtyValue

For comprehensive validation documentation, see: User Guide: Validation

---

pyvider.cty.validation

Classes

RecursionContext dataclass

RecursionContext(
    validation_graph: dict[int, ValidationNode] = dict(),
    validation_path: list[str] = list(),
    max_depth_reached: int = 0,
    total_validations: int = 0,
    validation_start_time: float = time.time(),
    max_depth_allowed: int = MAX_VALIDATION_DEPTH,
    max_object_revisits: int = MAX_OBJECT_REVISITS,
    max_validation_time_ms: int = MAX_VALIDATION_TIME_MS,
    validation_stopped: bool = False,
)

Thread-local context for tracking validation recursion.

Functions

reset

reset() -> None

Reset context for new validation session.

Source code in pyvider/cty/validation/recursion.py

def reset(self) -> None:
    """Reset context for new validation session."""
    self.validation_graph.clear()
    self.validation_path.clear()
    self.max_depth_reached = 0
    self.total_validations = 0
    self.validation_start_time = time.time()
    self.validation_stopped = False

RecursionDetector

RecursionDetector(context: RecursionContext | None = None)

Advanced recursion detector for CTY validation.

This detector uses sophisticated algorithms to distinguish between: - Circular references (object A -> object B -> object A) - Deep but finite nesting (legitimate complex configurations) - Performance pathological cases (excessive validation time)

Source code in pyvider/cty/validation/recursion.py

def __init__(self, context: RecursionContext | None = None) -> None:
    self.context = context or get_recursion_context()

Functions

should_continue_validation

should_continue_validation(
    value: Any, current_path: str = ""
) -> tuple[bool, str | None]

Determine if validation should continue for the given value.

Returns:

Type	Description
tuple[bool, str | None]	(should_continue, reason_if_stopped)

Production requirements: - Must handle legitimate deep nesting (1000+ levels) - Must detect genuine circular references quickly - Must provide detailed diagnostics for debugging - Must have predictable performance characteristics

Source code in pyvider/cty/validation/recursion.py

def should_continue_validation(self, value: Any, current_path: str = "") -> tuple[bool, str | None]:
    """
    Determine if validation should continue for the given value.

    Returns:
        (should_continue, reason_if_stopped)

    Production requirements:
    - Must handle legitimate deep nesting (1000+ levels)
    - Must detect genuine circular references quickly
    - Must provide detailed diagnostics for debugging
    - Must have predictable performance characteristics
    """

    # Performance safeguards - prevent pathological cases
    elapsed_ms = (time.time() - self.context.validation_start_time) * 1000
    if elapsed_ms > self.context.max_validation_time_ms:
        reason = (
            f"Validation timeout after {elapsed_ms:.1f}ms (max: {self.context.max_validation_time_ms}ms)"
        )
        logger.warning(
            "CTY validation timeout exceeded",
            elapsed_ms=elapsed_ms,
            max_allowed_ms=self.context.max_validation_time_ms,
            path=current_path,
            trace="advanced_recursion_detection",
        )
        return False, reason

    # Update context
    self.context.total_validations += 1
    current_depth = len(self.context.validation_path)
    self.context.max_depth_reached = max(self.context.max_depth_reached, current_depth)

    # Depth safeguards - only trigger for truly deep recursion
    if current_depth > self.context.max_depth_allowed:
        reason = f"Maximum nesting depth exceeded: {current_depth} > {self.context.max_depth_allowed}"
        logger.warning(
            "CTY validation depth limit exceeded",
            current_depth=current_depth,
            max_allowed=self.context.max_depth_allowed,
            path=current_path,
            trace="advanced_recursion_detection",
        )
        return False, reason

    # Skip cycle detection for primitive types and simple collections (performance optimization)
    if isinstance(value, (str, int, float, bool, type(None))):
        return True, None

    # Skip cycle detection for simple lists of primitives
    if isinstance(value, list) and all(
        isinstance(item, (str, int, float, bool, type(None))) for item in value
    ):
        return True, None

    # Sophisticated cycle detection
    value_id = id(value)
    value_type = type(value).__name__

    if value_id in self.context.validation_graph:
        node = self.context.validation_graph[value_id]
        node.visits += 1

        # Allow some revisits for complex but legitimate schemas
        if node.visits > self.context.max_object_revisits:
            reason = (
                f"Circular reference detected: {value_type} object visited "
                f"{node.visits} times (max: {self.context.max_object_revisits})"
            )
            logger.debug(
                "CTY circular reference detected",
                object_type=value_type,
                object_id=value_id,
                visits=node.visits,
                first_seen_depth=node.depth,
                current_depth=current_depth,
                path=current_path,
                trace="advanced_recursion_detection",
            )
            return False, reason

        # Log revisit for monitoring
        logger.debug(
            "CTY object revisited",
            object_type=value_type,
            object_id=value_id,
            visits=node.visits,
            path=current_path,
            trace="advanced_recursion_detection",
        )
    else:
        # First time seeing this object
        self.context.validation_graph[value_id] = ValidationNode(
            object_id=value_id,
            object_type=value_type,
            depth=current_depth,
            parent_path=current_path,
        )

    return True, None

enter_validation_scope

enter_validation_scope(scope_name: str) -> None

Enter a new validation scope for path tracking.

Source code in pyvider/cty/validation/recursion.py

def enter_validation_scope(self, scope_name: str) -> None:
    """Enter a new validation scope for path tracking."""
    self.context.validation_path.append(scope_name)

exit_validation_scope

exit_validation_scope() -> None

Exit the current validation scope.

Source code in pyvider/cty/validation/recursion.py

def exit_validation_scope(self) -> None:
    """Exit the current validation scope."""
    if self.context.validation_path:
        self.context.validation_path.pop()

get_current_path

get_current_path() -> str

Get the current validation path for diagnostics.

Source code in pyvider/cty/validation/recursion.py

def get_current_path(self) -> str:
    """Get the current validation path for diagnostics."""
    return " -> ".join(self.context.validation_path)

get_performance_metrics

get_performance_metrics() -> dict[str, Any]

Get performance metrics for monitoring and debugging.

Source code in pyvider/cty/validation/recursion.py

def get_performance_metrics(self) -> dict[str, Any]:
    """Get performance metrics for monitoring and debugging."""
    elapsed_ms = (time.time() - self.context.validation_start_time) * 1000
    return {
        "total_validations": self.context.total_validations,
        "max_depth_reached": self.context.max_depth_reached,
        "elapsed_ms": elapsed_ms,
        "objects_in_graph": len(self.context.validation_graph),
        "avg_validations_per_ms": self.context.total_validations / max(elapsed_ms, 0.001),
        "current_path": self.get_current_path(),
    }

Functions

clear_recursion_context

clear_recursion_context() -> None

Clear thread-local recursion context.

Source code in pyvider/cty/validation/recursion.py

def clear_recursion_context() -> None:
    """Clear thread-local recursion context."""
    if hasattr(_thread_local, "recursion_context"):
        _thread_local.recursion_context.reset()

get_recursion_context

get_recursion_context() -> RecursionContext

Get or create thread-local recursion context.

Source code in pyvider/cty/validation/recursion.py

def get_recursion_context() -> RecursionContext:
    """Get or create thread-local recursion context."""
    if not hasattr(_thread_local, "recursion_context"):
        _thread_local.recursion_context = RecursionContext()
    return cast(RecursionContext, _thread_local.recursion_context)

with_recursion_detection

with_recursion_detection(
    func: Callable[..., Any],
) -> Callable[..., Any]

Decorator for advanced recursion detection in validation functions.

Source code in pyvider/cty/validation/recursion.py

def with_recursion_detection(func: Callable[..., Any]) -> Callable[..., Any]:
    """
    Decorator for advanced recursion detection in validation functions.
    """

    @wraps(func)
    def wrapper(self: Any, value: Any, *args: Any, **kwargs: Any) -> Any:
        context = get_recursion_context()
        # A call is top-level if the context has not been used yet.
        is_top_level_call = context.total_validations == 0
        if is_top_level_call:
            context.reset()

        detector = RecursionDetector(context)
        scope_name = f"{self.__class__.__name__}.validate(type={type(value).__name__})"

        with error_boundary(
            context={
                "operation": "recursion_detection",
                "type_name": self.__class__.__name__,
                "value_type": type(value).__name__,
                "validation_depth": len(context.validation_path),
                "total_validations": context.total_validations,
            }
        ):
            detector.enter_validation_scope(scope_name)

            try:
                # Check if validation was already stopped by a nested call
                if context.validation_stopped:
                    from pyvider.cty.values import CtyValue

                    return CtyValue.unknown(self)

                should_continue, reason = detector.should_continue_validation(
                    value, detector.get_current_path()
                )
                if not should_continue:
                    from pyvider.cty.values import CtyValue

                    # Set flag to stop all parent validations
                    context.validation_stopped = True

                    logger.warning(
                        "CTY validation stopped due to recursion detection",
                        reason=reason,
                        value_type=type(value).__name__,
                        path=detector.get_current_path(),
                    )
                    return CtyValue.unknown(self)

                # The decorator no longer passes the internal flag down.
                result = func(self, value, *args, **kwargs)

                # Check again after validation in case a nested call stopped validation
                if context.validation_stopped:
                    from pyvider.cty.values import CtyValue

                    return CtyValue.unknown(self)

                return result
            finally:
                detector.exit_validation_scope()

    return wrapper

validate_config

validate_config(schema: Any, config: Any) -> None

Validates a configuration against a CtyType schema.

This function serves as the primary entry point for validation, delegating to the validate method of the provided schema. It allows the CtyValidationError to propagate, which is the expected contract for testing and low-level framework integration.

Parameters:

Name	Type	Description	Default
schema	Any	The CtyType object to validate against.	required
config	Any	The raw Python data to validate.	required

Raises:

Type	Description
CtyValidationError	If the configuration does not conform to the schema.

Source code in pyvider/cty/validation/__init__.py

def validate_config(schema: Any, config: Any) -> None:
    """
    Validates a configuration against a CtyType schema.

    This function serves as the primary entry point for validation,
    delegating to the `validate` method of the provided schema. It allows
    the CtyValidationError to propagate, which is the expected contract
    for testing and low-level framework integration.

    Args:
        schema: The CtyType object to validate against.
        config: The raw Python data to validate.

    Raises:
        CtyValidationError: If the configuration does not conform to the schema.
    """
    # The schema (a CtyType instance) has the validation logic.
    # We simply call it and let it raise its exception on failure.
    schema.validate(config)