Index

pyvider.cty

TODO: Add module docstring.

Attributes

BytesCapsule module-attribute

BytesCapsule = CtyCapsule('Bytes', bytes)

A capsule type for wrapping raw bytes.

Classes

CtyCapsule

CtyCapsule(capsule_name: str, py_type: type)

Bases: CtyType[Any]

Represents a capsule type in the Cty type system. Capsule types are opaque types that can be used to wrap arbitrary Python objects.

Source code in pyvider/cty/types/capsule.py

def __init__(self, capsule_name: str, py_type: type) -> None:
    super().__init__()
    self.name = capsule_name
    self._py_type = py_type

CtyCapsuleWithOps

CtyCapsuleWithOps(
    capsule_name: str,
    py_type: type,
    *,
    equal_fn: Callable[[Any, Any], bool] | None = None,
    hash_fn: Callable[[Any], int] | None = None,
    convert_fn: (
        Callable[[Any, CtyType[Any]], CtyValue[Any] | None]
        | None
    ) = None
)

Bases: CtyCapsule

A CtyCapsule that supports custom operations like equality, hashing, and conversion.

Initializes a CtyCapsule with custom operational functions.

Source code in pyvider/cty/types/capsule.py

def __init__(
    self,
    capsule_name: str,
    py_type: type,
    *,
    equal_fn: Callable[[Any, Any], bool] | None = None,
    hash_fn: Callable[[Any], int] | None = None,
    convert_fn: Callable[[Any, CtyType[Any]], CtyValue[Any] | None] | None = None,
) -> None:
    """
    Initializes a CtyCapsule with custom operational functions.
    """
    super().__init__(capsule_name, py_type)
    self.equal_fn = equal_fn
    self.hash_fn = hash_fn
    self.convert_fn = convert_fn
    self._validate_ops_arity()

Functions

CtyConversionError

CtyConversionError(
    message: str,
    *,
    source_value: object | None = None,
    target_type: object | None = None,
    **kwargs: Any
)

Bases: CtyError

Base for CTY value or type conversion errors.

Initializes the CtyConversionError.

Parameters:

Name	Type	Description	Default
message	str	The base error message.	required
source_value	object | None	The value that was being converted.	None
target_type	object | None	The intended target type of the conversion.	None
**kwargs	Any	Additional foundation error context.	{}

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

def __init__(
    self,
    message: str,
    *,
    source_value: object | None = None,
    target_type: object | None = None,
    **kwargs: Any,
) -> None:
    """
    Initializes the CtyConversionError.

    Args:
        message: The base error message.
        source_value: The value that was being converted.
        target_type: The intended target type of the conversion.
        **kwargs: Additional foundation error context.
    """
    self.source_value = source_value
    self.target_type = target_type

    # Add rich conversion context
    context = kwargs.setdefault("context", {})
    context["cty.operation"] = "conversion"
    context["cty.error_category"] = "type_conversion"

    # Build message with old format for compatibility
    context_parts = []
    if source_value is not None:
        context_parts.append(f"source_type={type(source_value).__name__}")
        # Also add to foundation context
        context["conversion.source_type"] = type(source_value).__name__
        context["conversion.source_value_type"] = type(source_value).__name__

        # Add value analysis for better debugging
        if hasattr(source_value, "type") and hasattr(source_value, "is_null"):
            context["conversion.source_cty_type"] = str(source_value.type)
            context["conversion.source_is_null"] = source_value.is_null
            if hasattr(source_value, "is_unknown"):
                context["conversion.source_is_unknown"] = source_value.is_unknown

    if target_type is not None:
        target_name = target_type.__name__ if hasattr(target_type, "__name__") else str(target_type)
        context_parts.append(f"target_type={target_name}")
        context["conversion.target_type"] = target_name
        context["conversion.target_type_str"] = str(target_type)

    if context_parts:
        message = f"{message} ({', '.join(context_parts)})"

    super().__init__(message, **kwargs)

Functions

CtyDynamic

Bases: CtyType[object]

Represents a dynamic type that can hold any CtyValue.

Functions

validate

validate(value: object) -> CtyValue[Any]

Validates a raw Python value for a dynamic type. The result is always a CtyValue of type CtyDynamic, which wraps the inferred concrete value.

Source code in pyvider/cty/types/structural/dynamic.py

@with_recursion_detection
def validate(self, value: object) -> CtyValue[Any]:
    """
    Validates a raw Python value for a dynamic type. The result is always a
    CtyValue of type CtyDynamic, which wraps the inferred concrete value.
    """
    from pyvider.cty.conversion.raw_to_cty import infer_cty_type_from_raw
    from pyvider.cty.parser import parse_tf_type_to_ctytype
    from pyvider.cty.values import CtyValue

    if isinstance(value, CtyValue):
        if isinstance(value.type, CtyDynamic):
            return cast(CtyValue[Any], value)  # type: ignore[redundant-cast]
        return CtyValue(vtype=self, value=value)

    if value is None:
        return CtyValue.null(self)

    if isinstance(value, list) and len(value) == 2 and isinstance(value[0], bytes):
        try:
            type_spec = json.loads(value[0].decode("utf-8"))
            actual_type = parse_tf_type_to_ctytype(type_spec)
            concrete_value = actual_type.validate(value[1])
            return CtyValue(vtype=self, value=concrete_value)
        except json.JSONDecodeError as e:
            raise DeserializationError(
                "Failed to decode dynamic value type spec from JSON during validation"
            ) from e
        except CtyValidationError as e:
            raise e

    inferred_type = infer_cty_type_from_raw(value)
    concrete_value = inferred_type.validate(value)
    return CtyValue(vtype=self, value=concrete_value)

CtyMark

Represents a mark that can be applied to a cty.Value. The 'details' attribute is automatically converted to a hashable frozenset.

CtyType

Bases: CtyTypeProtocol[T], Generic[T], ABC

Generic abstract base class for all Cty types.

Functions

is_dynamic_type

is_dynamic_type() -> bool

Returns True if this type is CtyDynamic.

Source code in pyvider/cty/types/base.py

def is_dynamic_type(self) -> bool:
    """Returns True if this type is CtyDynamic."""
    return False

CtyTypeParseError

CtyTypeParseError(
    message: str, type_string: str, **kwargs: Any
)

Bases: CtyConversionError

Raised when a CTY type string cannot be parsed.

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

def __init__(self, message: str, type_string: str, **kwargs: Any) -> None:
    self.type_string = type_string

    # Add parsing context
    context = kwargs.setdefault("context", {})
    context["cty.conversion_category"] = "type_parsing"
    context["cty.parse_input"] = str(type_string)[:100]  # Truncate for safety
    context["cty.parse_input_type"] = type(type_string).__name__

    full_message = f"{message}: '{type_string}'"
    super().__init__(full_message, source_value=type_string, **kwargs)

CtyValidationError

CtyValidationError(
    message: str,
    value: object = None,
    type_name: str | None = None,
    path: CtyPath | None = None,
    **kwargs: Any
)

Bases: ValidationError

Base exception for all validation errors.

Inherits from foundation's ValidationError for enhanced diagnostics and automatic retry/circuit breaker support where applicable.

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

def __init__(
    self,
    message: str,
    value: object = None,
    type_name: str | None = None,
    path: CtyPath | None = None,
    **kwargs: Any,
) -> None:
    self.value = value
    self.type_name = type_name
    self.path = path
    self.message = message

    # Add rich context to foundation error with more detailed information
    context = kwargs.setdefault("context", {})

    # Core CTY context
    if type_name:
        context["cty.type"] = type_name
    if path:
        context["cty.path"] = str(path)
        context["cty.path_depth"] = len(path.steps) if path else 0

    # Value context with safe representation
    if value is not None:
        context["cty.value_type"] = type(value).__name__
        # Safe value representation for debugging (truncated to avoid huge objects)
        try:
            value_repr = repr(value)
            context["cty.value_repr"] = value_repr[:200] + "..." if len(value_repr) > 200 else value_repr
        except Exception:
            context["cty.value_repr"] = f"<repr failed for {type(value).__name__}>"

    # Add validation context if available
    context["cty.validation_stage"] = "type_validation"

    super().__init__(self.message, **kwargs)

Functions

__str__

__str__() -> str

Creates a user-friendly, path-aware error message.

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

def __str__(self) -> str:
    """Creates a user-friendly, path-aware error message."""
    path_str = str(self.path) if self.path and self.path.steps else ""
    core_message = self.message

    if path_str and path_str != "(root)":
        return f"At {path_str}: {core_message}"

    return core_message

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,
        )

parse_tf_type_to_ctytype

parse_tf_type_to_ctytype(tf_type: Any) -> CtyType[Any]

Parses a Terraform type constraint, represented as a raw Python object (typically from JSON), into a CtyType instance.

Source code in pyvider/cty/parser.py

def parse_tf_type_to_ctytype(tf_type: Any) -> CtyType[Any]:  # noqa: C901
    """
    Parses a Terraform type constraint, represented as a raw Python object
    (typically from JSON), into a CtyType instance.
    """
    with error_boundary(
        context={
            "operation": "terraform_type_parsing",
            "tf_type": str(tf_type),
            "tf_type_python_type": type(tf_type).__name__,
        }
    ):
        if isinstance(tf_type, str):
            match tf_type:
                case "string":
                    return CtyString()
                case "number":
                    return CtyNumber()
                case "bool":
                    return CtyBool()
                case "dynamic":
                    return CtyDynamic()
                case _:
                    raise CtyValidationError(f"Unknown primitive type name: '{tf_type}'")

        if isinstance(tf_type, list) and len(tf_type) == 2:
            type_kind, type_spec = tf_type

            # Handle collection types where the spec is a single type
            if type_kind in (TYPE_KIND_LIST, TYPE_KIND_SET, TYPE_KIND_MAP):
                element_type = parse_tf_type_to_ctytype(type_spec)
                match type_kind:
                    case "list":
                        return CtyList(element_type=element_type)
                    case "set":
                        return CtySet(element_type=element_type)
                    case "map":
                        return CtyMap(element_type=element_type)

            # Handle structural types where the spec is a container
            match type_kind:
                case "object":
                    if not isinstance(type_spec, dict):
                        raise CtyValidationError(
                            f"Object type spec must be a dictionary, got {type(type_spec).__name__}"
                        )
                    attr_types = {name: parse_tf_type_to_ctytype(spec) for name, spec in type_spec.items()}
                    return CtyObject(attribute_types=attr_types)
                case "tuple":
                    if not isinstance(type_spec, list):
                        raise CtyValidationError(
                            f"Tuple type spec must be a list, got {type(type_spec).__name__}"
                        )
                    elem_types = tuple(parse_tf_type_to_ctytype(spec) for spec in type_spec)
                    return CtyTuple(element_types=elem_types)

        raise CtyValidationError(f"Invalid Terraform type specification: {tf_type}")

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))