nat.observability.mixin.type_introspection_mixin

Attributes

logger

Classes

TypeIntrospectionMixin

Hybrid mixin class providing type introspection capabilities for generic classes.

Module Contents

logger

class TypeIntrospectionMixin

Hybrid mixin class providing type introspection capabilities for generic classes.

This mixin combines the DecomposedType class utilities with MRO traversal to properly handle complex inheritance chains like HeaderRedactionProcessor or ProcessingExporter.

_extract_types_from_signature_method() → tuple[type[Any], type[Any]] | None

Extract input/output types from the signature method.

This method looks for a signature method (either defined via _signature_method class attribute or discovered generically) and extracts input/output types from its method signature.

Returns:

tuple[type[Any], type[Any]] | None: (input_type, output_type) or None if not found.

_discover_signature_method() → str | None

Discover any method suitable for type introspection.

Looks for any method with the signature pattern: method(self, param: Type) -> ReturnType Any method matching this pattern is functionally equivalent for type introspection purposes.

Returns:

str | None: Method name or None if not found

_resolve_typevar_recursively(

type_arg: Any,

typevar_mapping: dict[TypeVar, type[Any]],

) → Any

Recursively resolve TypeVars within complex types.

Args:

type_arg (Any): The type argument to resolve (could be a TypeVar, generic type, etc.) typevar_mapping (dict[TypeVar, type[Any]]): Current mapping of TypeVars to concrete types

Returns:

Any: The resolved type with all TypeVars substituted

_contains_typevar(type_arg: Any) → bool

Check if a type contains any TypeVars (including nested ones).

Args:

type_arg (Any): The type to check

Returns:

bool: True if the type contains any TypeVars

_build_typevar_mapping() → dict[TypeVar, type[Any]]

Build TypeVar to concrete type mapping from MRO traversal.

Returns:

dict[TypeVar, type[Any]]: Mapping of TypeVars to concrete types

_extract_instance_types_from_mro() → tuple[type[Any], type[Any]] | None

Extract Generic[InputT, OutputT] types by traversing the MRO.

This handles complex inheritance chains by looking for the base class and resolving TypeVars through the inheritance hierarchy.

Returns:

tuple[type[Any], type[Any]] | None: (input_type, output_type) or None if not found

_extract_input_output_types() → tuple[type[Any], type[Any]]

Extract both input and output types using available approaches.

Returns:

tuple[type[Any], type[Any]]: (input_type, output_type)

Raises:

ValueError: If types cannot be extracted

property input_type: type[Any]

Get the input type of the instance.

Returns:

type[Any]: The input type

property output_type: type[Any]

Get the output type of the instance.

Returns:

type[Any]: The output type

_get_union_info(

type_obj: type[Any],

) → tuple[bool, tuple[type, Ellipsis] | None]

Get union information for a type.

Args:

type_obj (type[Any]): The type to analyze

Returns:

tuple[bool, tuple[type, …] | None]: (is_union, union_types_or_none)

property has_union_input: bool

Check if the input type is a union type.

Returns:

bool: True if the input type is a union type, False otherwise

property has_union_output: bool

Check if the output type is a union type.

Returns:

bool: True if the output type is a union type, False otherwise

property input_union_types: tuple[type, Ellipsis] | None

Get the individual types in an input union.

Returns:

tuple[type, …] | None: The individual types in an input union or None if not found

property output_union_types: tuple[type, Ellipsis] | None

Get the individual types in an output union.

Returns:

tuple[type, …] | None: The individual types in an output union or None if not found

is_compatible_with_input(source_type: type) → bool

Check if a source type is compatible with this instance’s input type.

Uses Pydantic-based type compatibility checking for strict type matching. This focuses on proper type relationships rather than batch compatibility.

Args:

source_type (type): The source type to check

Returns:

bool: True if the source type is compatible with the input type, False otherwise

is_output_compatible_with(target_type: type) → bool

Check if this instance’s output type is compatible with a target type.

Uses Pydantic-based type compatibility checking for strict type matching. This focuses on proper type relationships rather than batch compatibility.

Args:

target_type (type): The target type to check

Returns:

bool: True if the output type is compatible with the target type, False otherwise

_is_pydantic_type_compatible(

source_type: type,

target_type: type,

) → bool

Check strict type compatibility without batch compatibility hacks.

This focuses on proper type relationships: exact matches and subclass relationships.

Args:

source_type (type): The source type to check target_type (type): The target type to check compatibility with

Returns:

bool: True if types are compatible, False otherwise

_get_input_validator() → type[pydantic.BaseModel]

Create a Pydantic model for validating input types.

Returns:

type[BaseModel]: The Pydantic model for validating input types

_get_output_validator() → type[pydantic.BaseModel]

Create a Pydantic model for validating output types.

Returns:

type[BaseModel]: The Pydantic model for validating output types

validate_input_type(item: Any) → bool

Validate that an item matches the expected input type using Pydantic.

Args:

item (Any): The item to validate

Returns:

bool: True if the item matches the input type, False otherwise

validate_output_type(item: Any) → bool

Validate that an item matches the expected output type using Pydantic.

Args:

item (Any): The item to validate

Returns:

bool: True if the item matches the output type, False otherwise

extract_non_optional_type(type_obj: type | types.UnionType) → Any

Extract the non-None type from Optional[T] or Union[T, None] types.

This is useful when you need to pass a type to a system that doesn’t understand Optional types (like registries that expect concrete types).

Args:

type_obj (type | types.UnionType): The type to extract from (could be Optional[T] or Union[T, None])

Returns:

Any: The actual type without None, or the original type if not a union with None