Index

pyvider.cty.conversion

TODO: Add module docstring.

Functions

convert

convert(
    value: CtyValue[Any], target_type: CtyType[Any]
) -> CtyValue[Any]

Converts a CtyValue to a new CtyValue of the target CtyType.

Source code in pyvider/cty/conversion/explicit.py

def convert(value: CtyValue[Any], target_type: CtyType[Any]) -> CtyValue[Any]:  # noqa: C901
    """
    Converts a CtyValue to a new CtyValue of the target CtyType.
    """
    with error_boundary(
        context={
            "operation": "cty_value_conversion",
            "source_type": str(value.type),
            "target_type": str(target_type),
            "value_is_null": value.is_null,
            "value_is_unknown": value.is_unknown,
        }
    ):
        # Early exit cases
        if value.type.equal(target_type):
            return value

        if value.is_null:
            return CtyValue.null(target_type)
        if value.is_unknown:
            return CtyValue.unknown(target_type)

        # Capsule conversion with operations
        if isinstance(value.type, CtyCapsuleWithOps) and value.type.convert_fn:
            result = value.type.convert_fn(value.value, target_type)
            if result is None:
                error_message = ERR_CAPSULE_CANNOT_CONVERT.format(
                    value_type=value.type, target_type=target_type
                )
                raise CtyConversionError(
                    error_message,
                    source_value=value,
                    target_type=target_type,
                )
            if not isinstance(result, CtyValue):
                error_message = ERR_CUSTOM_CONVERTER_NON_CTYVALUE
                raise CtyConversionError(
                    error_message,
                    source_value=value,
                    target_type=target_type,
                )
            if not result.type.equal(target_type):
                error_message = ERR_CUSTOM_CONVERTER_WRONG_TYPE.format(
                    result_type=result.type, target_type=target_type
                )
                raise CtyConversionError(
                    error_message,
                    source_value=value,
                    target_type=target_type,
                )
            return result.with_marks(set(value.marks))

        # Dynamic type handling
        if isinstance(value.type, CtyDynamic):
            if not isinstance(value.value, CtyValue):
                error_message = ERR_DYNAMIC_VALUE_NOT_CTYVALUE
                raise CtyConversionError(error_message, source_value=value)
            return convert(value.value, target_type)

        if isinstance(target_type, CtyDynamic):
            return value.with_marks(set(value.marks))

        # String conversion
        if isinstance(target_type, CtyString) and not isinstance(value.type, CtyCapsule):
            raw = value.value
            new_val = ("true" if raw else "false") if isinstance(raw, bool) else str(raw)
            return CtyValue(target_type, new_val).with_marks(set(value.marks))

        # Number conversion
        if isinstance(target_type, CtyNumber):
            try:
                validated = target_type.validate(value.value)
                return validated.with_marks(set(value.marks))
            except CtyValidationError as e:
                error_message = ERR_CANNOT_CONVERT_VALIDATION.format(
                    value_type=value.type, target_type=target_type, message=e.message
                )
                raise CtyConversionError(
                    error_message,
                    source_value=value,
                    target_type=target_type,
                ) from e

        # Boolean conversion
        if isinstance(target_type, CtyBool):
            if isinstance(value.type, CtyString):
                s = str(value.value).lower()
                if s == "true":
                    return CtyValue(target_type, True).with_marks(set(value.marks))
                if s == "false":
                    return CtyValue(target_type, False).with_marks(set(value.marks))
            error_message = ERR_CANNOT_CONVERT_TO_BOOL.format(value_type=value.type)
            raise CtyConversionError(
                error_message,
                source_value=value,
                target_type=target_type,
            )

        # Collection conversions
        if isinstance(target_type, CtySet) and isinstance(value.type, CtyList | CtyTuple):
            converted: CtyValue[Any] = target_type.validate(value.value).with_marks(set(value.marks))
            return converted

        if isinstance(target_type, CtyList) and isinstance(value.type, CtySet | CtyTuple):
            converted = target_type.validate(value.value).with_marks(set(value.marks))
            return converted

        if isinstance(target_type, CtyList) and isinstance(value.type, CtyList):
            if target_type.element_type.equal(value.type.element_type):
                return value
            if isinstance(target_type.element_type, CtyDynamic):
                converted = target_type.validate(value.value).with_marks(set(value.marks))
                return converted

        # Object conversion
        if isinstance(target_type, CtyObject) and isinstance(value.type, CtyObject):
            new_attrs = {}
            source_attrs = value.value
            if not isinstance(source_attrs, dict):
                error_message = ERR_SOURCE_OBJECT_NOT_DICT
                raise CtyConversionError(error_message)
            source_attrs_dict = cast(dict[str, CtyValue[Any]], source_attrs)
            for name, target_attr_type in target_type.attribute_types.items():
                if name in source_attrs_dict:
                    new_attrs[name] = convert(source_attrs_dict[name], target_attr_type)
                elif name in target_type.optional_attributes:
                    new_attrs[name] = CtyValue.null(target_attr_type)
                else:
                    error_message = ERR_MISSING_REQUIRED_ATTRIBUTE.format(name=name)
                    raise CtyConversionError(error_message)
            converted = target_type.validate(new_attrs).with_marks(set(value.marks))
            return converted

        # Fallback - no conversion available
        error_message = ERR_CANNOT_CONVERT_GENERAL.format(value_type=value.type, target_type=target_type)
        raise CtyConversionError(
            error_message,
            source_value=value,
            target_type=target_type,
        )

cty_to_native

cty_to_native(value: CtyValue[Any] | Any) -> Any

Converts a CtyValue to its raw Python representation using an iterative approach to avoid recursion limits. This is safe for deeply nested structures.

Source code in pyvider/cty/conversion/adapter.py

def cty_to_native(value: CtyValue[Any] | Any) -> Any:  # noqa: C901
    """
    Converts a CtyValue to its raw Python representation using an iterative
    approach to avoid recursion limits. This is safe for deeply nested structures.
    """
    if not isinstance(value, CtyValue):
        return value

    if value.is_unknown:
        return None  # Gracefully handle unknown values by returning None.

    POST_PROCESS = object()
    work_stack: list[Any] = [value]
    results: dict[int, Any] = {}
    processing: set[int] = set()

    while work_stack:
        current_item = work_stack.pop()

        if current_item is POST_PROCESS:
            val_to_process = work_stack.pop()
            val_id = id(val_to_process)
            processing.remove(val_id)

            # Robustness check for malformed collection values
            if not hasattr(val_to_process.value, "__iter__"):
                if isinstance(val_to_process.type, CtyList | CtySet):
                    results[val_id] = []
                elif isinstance(val_to_process.type, CtyTuple):
                    results[val_id] = ()
                elif isinstance(val_to_process.type, CtyMap | CtyObject):
                    results[val_id] = {}
                continue

            if isinstance(val_to_process.type, CtyDynamic):
                inner_id = id(val_to_process.value)
                results[val_id] = results[inner_id]
            elif isinstance(val_to_process.type, CtyObject | CtyMap):
                dict_val = cast(dict[str, Any], val_to_process.value)
                results[val_id] = {k: results[id(v)] for k, v in dict_val.items()}
            elif isinstance(val_to_process.type, CtyList):
                list_val = cast(list[Any], val_to_process.value)
                results[val_id] = [results[id(item)] for item in list_val]
            elif isinstance(val_to_process.type, CtySet):
                # Use _canonical_sort_key for consistent sorting of set elements
                set_val = cast(set[Any], val_to_process.value)
                results[val_id] = sorted(
                    [results[id(item)] for item in set_val],
                    key=lambda v: v._canonical_sort_key() if isinstance(v, CtyValue) else repr(v),
                )
            elif isinstance(val_to_process.type, CtyTuple):
                tuple_val = cast(tuple[Any, ...], val_to_process.value)
                results[val_id] = tuple(results[id(item)] for item in tuple_val)
            continue

        if not isinstance(current_item, CtyValue):
            results[id(current_item)] = current_item
            continue

        if current_item.is_unknown:
            results[id(current_item)] = None
            continue
        if current_item.is_null:
            results[id(current_item)] = None
            continue

        item_id = id(current_item)
        if item_id in results or item_id in processing:
            continue

        if isinstance(
            current_item.type,
            CtyObject | CtyMap | CtyList | CtySet | CtyTuple | CtyDynamic,
        ):
            processing.add(item_id)
            work_stack.extend([current_item, POST_PROCESS])

            if isinstance(current_item.type, CtyDynamic):
                work_stack.append(current_item.value)
            elif hasattr(current_item.value, "__iter__"):  # Robustness check
                if isinstance(current_item.value, dict):
                    dict_val = cast(dict[str, Any], current_item.value)
                    child_values = list(dict_val.values())
                else:
                    iterable_val = cast(list[Any] | set[Any] | tuple[Any, ...], current_item.value)
                    child_values = list(iterable_val)
                work_stack.extend(reversed(child_values))
        else:
            inner_val = current_item.value
            if isinstance(inner_val, Decimal):
                exponent = inner_val.as_tuple().exponent
                if isinstance(exponent, int) and exponent >= 0:
                    results[item_id] = int(inner_val)
                else:
                    results[item_id] = float(inner_val)
            else:
                results[item_id] = inner_val

    return results.get(id(value))

encode_cty_type_to_wire_json

encode_cty_type_to_wire_json(cty_type: CtyType[Any]) -> Any

Encodes a CtyType into a JSON-serializable structure for the wire format by delegating to the type's own _to_wire_json method.

Source code in pyvider/cty/conversion/type_encoder.py

def encode_cty_type_to_wire_json(cty_type: CtyType[Any]) -> Any:
    """
    Encodes a CtyType into a JSON-serializable structure for the wire format
    by delegating to the type's own `_to_wire_json` method.
    """
    if not isinstance(cty_type, CtyType):
        error_message = ERR_EXPECTED_CTYTYPE.format(type_name=type(cty_type).__name__)
        raise TypeError(error_message)
    return cty_type._to_wire_json()

infer_cty_type_from_raw

infer_cty_type_from_raw(value: Any) -> CtyType[Any]

Infers the most specific CtyType from a raw Python value. This function uses an iterative approach with a work stack to avoid recursion limits and leverages a context-aware cache for performance and thread-safety.

Source code in pyvider/cty/conversion/raw_to_cty.py

@with_inference_cache
def infer_cty_type_from_raw(value: Any) -> CtyType[Any]:  # noqa: C901
    """
    Infers the most specific CtyType from a raw Python value.
    This function uses an iterative approach with a work stack to avoid recursion limits
    and leverages a context-aware cache for performance and thread-safety.
    """
    with error_boundary(
        context={
            "operation": "cty_type_inference",
            "value_type": type(value).__name__,
            "is_attrs_class": attrs.has(type(value)) if hasattr(value, "__class__") else False,
            "value_repr": str(value)[:100] if value is not None else "None",  # Truncated for safety
        }
    ):
        from pyvider.cty.types import (
            CtyBool,
            CtyDynamic,
            CtyList,
            CtyMap,
            CtyNumber,
            CtyObject,
            CtySet,
            CtyString,
            CtyTuple,
            CtyType,
        )

        if isinstance(value, CtyValue) or value is None:
            return CtyDynamic()

        if isinstance(value, CtyType):
            return CtyDynamic()

        if attrs.has(type(value)):
            value = _attrs_to_dict_safe(value)

    container_cache = get_container_schema_cache()

    # If no cache is available (e.g., in worker threads for thread safety),
    # proceed without caching
    structural_key = None
    if container_cache is not None:
        structural_key = _get_structural_cache_key(value)
        if structural_key in container_cache:
            return container_cache[structural_key]

    POST_PROCESS = object()
    work_stack: list[Any] = [value]
    results: dict[int, CtyType[Any]] = {}
    processing: set[int] = set()

    while work_stack:
        current_item = work_stack.pop()

        if current_item is POST_PROCESS:
            container = work_stack.pop()
            container_id = id(container)
            processing.remove(container_id)

            if isinstance(container, dict) and all(isinstance(k, str) for k in container):
                container = {unicodedata.normalize("NFC", k): v for k, v in container.items()}

            child_values = container.values() if isinstance(container, dict) else container
            child_types = [
                (v.type if isinstance(v, CtyValue) else results.get(id(v), CtyDynamic())) for v in child_values
            ]

            inferred_schema: CtyType[Any]
            if isinstance(container, dict):
                if not container:
                    inferred_schema = CtyObject({})
                elif not all(isinstance(k, str) for k in container):
                    unified = _unify_types(set(child_types))
                    inferred_schema = CtyMap(element_type=unified)
                else:
                    attr_types = dict(zip(container.keys(), child_types, strict=True))
                    inferred_schema = CtyObject(attribute_types=attr_types)
            elif isinstance(container, tuple):
                inferred_schema = CtyTuple(element_types=tuple(child_types))
            elif isinstance(container, list | set):
                unified = _unify_types(set(child_types))
                inferred_schema = (
                    CtyList(element_type=unified)
                    if isinstance(container, list)
                    else CtySet(element_type=unified)
                )
            else:
                inferred_schema = CtyDynamic()

            results[container_id] = inferred_schema
            continue

        if attrs.has(type(current_item)) and not isinstance(current_item, CtyType):
            try:
                current_item = _attrs_to_dict_safe(current_item)
            except TypeError:
                results[id(current_item)] = CtyDynamic()
                continue

        if current_item is None:
            continue
        item_id = id(current_item)
        if item_id in results or item_id in processing:
            continue
        if isinstance(current_item, CtyValue):
            results[item_id] = current_item.type
            continue

        if not isinstance(current_item, dict | list | tuple | set):
            if isinstance(current_item, bool):
                results[item_id] = CtyBool()
            elif isinstance(current_item, int | float | Decimal):
                results[item_id] = CtyNumber()
            elif isinstance(current_item, str | bytes):
                results[item_id] = CtyString()
            else:
                results[item_id] = CtyDynamic()
            continue

        structural_key = _get_structural_cache_key(current_item)
        if container_cache is not None and structural_key in container_cache:
            results[item_id] = container_cache[structural_key]
            continue

        processing.add(item_id)
        work_stack.extend([current_item, POST_PROCESS])
        work_stack.extend(
            reversed(list(current_item.values() if isinstance(current_item, dict) else current_item))
        )

    final_type = results.get(id(value), CtyDynamic())

    # Cache the result if caching is available
    if container_cache is not None:
        final_structural_key = _get_structural_cache_key(value)
        container_cache[final_structural_key] = final_type

    return final_type

inference_cache_context

inference_cache_context() -> Generator[None]

Provide isolated inference caches for type inference operations.

Creates scoped caches that are automatically cleaned up when exiting the context. Nested contexts reuse the parent cache. Respects the configuration setting for enabling/disabling caches.

Yields:

Type	Description
Generator[None]	None (use get_*_cache() functions within context)

Examples:

>>> with inference_cache_context():
...     # Caches are active here
...     result = infer_cty_type_from_raw(data)
... # Caches automatically cleared

Source code in pyvider/cty/conversion/inference_cache.py

@contextmanager
def inference_cache_context() -> Generator[None]:
    """Provide isolated inference caches for type inference operations.

    Creates scoped caches that are automatically cleaned up when exiting
    the context. Nested contexts reuse the parent cache. Respects the
    configuration setting for enabling/disabling caches.

    Yields:
        None (use get_*_cache() functions within context)

    Examples:
        >>> with inference_cache_context():
        ...     # Caches are active here
        ...     result = infer_cty_type_from_raw(data)
        ... # Caches automatically cleared
    """
    from pyvider.cty.config.runtime import CtyConfig

    config = CtyConfig.get_current()
    if not config.enable_type_inference_cache:
        yield
        return

    with _structural_key_cache.scope(), _container_schema_cache.scope():
        yield

unify

unify(types: Iterable[CtyType[Any]]) -> CtyType[Any]

Finds a single common CtyType that all of the given types can convert to. This is a wrapper that enables caching by converting input to a frozenset.

Source code in pyvider/cty/conversion/explicit.py

def unify(types: Iterable[CtyType[Any]]) -> CtyType[Any]:
    """
    Finds a single common CtyType that all of the given types can convert to.
    This is a wrapper that enables caching by converting input to a frozenset.
    """
    return _unify_frozen(frozenset(types))

with_inference_cache

with_inference_cache(func: F) -> F

Decorator providing isolated inference cache for function execution.

Ensures thread/async safety by providing each invocation with its own cache context via ContextVar-based scoping.

Parameters:

Name	Type	Description	Default
func	F	Function to decorate	required

Returns:

Type	Description
F	Decorated function with cache context

Examples:

>>> @with_inference_cache
... def infer_type(value):
...     # Has access to inference caches
...     pass

Source code in pyvider/cty/conversion/inference_cache.py

def with_inference_cache(func: F) -> F:
    """Decorator providing isolated inference cache for function execution.

    Ensures thread/async safety by providing each invocation with its own
    cache context via ContextVar-based scoping.

    Args:
        func: Function to decorate

    Returns:
        Decorated function with cache context

    Examples:
        >>> @with_inference_cache
        ... def infer_type(value):
        ...     # Has access to inference caches
        ...     pass
    """

    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        with inference_cache_context():
            return func(*args, **kwargs)

    return wrapper  # type: ignore[return-value]