haive.core.schema.compatibility.utils¶

from typing import Any. Utility functions for the schema compatibility module.

Functions¶

`calculate_similarity`(str1, str2)	Calculate similarity between two strings (0-1).
`create_example_value`(type_hint)	Create an example value for a type hint.
`create_schema_diff`(schema1, schema2)	Create a diff between two schemas.
`estimate_memory_usage`(schema)	Estimate memory usage of a schema instance in bytes.
`extract_path_value`(data, path[, separator, default])	Extract value from nested dict using path.
`extract_type_name`(type_hint)	Extract a readable name from a type hint.
`find_similar_fields`(target_field, source_fields[, ...])	Find similar field names with scores.
`flatten_nested_dict`(data[, parent_key, separator])	Flatten nested dictionary.
`format_type_path`(types)	Format a type conversion path for display.
`generate_schema_hash`(schema)	Generate a hash for schema comparison.
`get_all_subclasses`(cls)	Get all subclasses of a class recursively.
`memoize`(func)	Simple memoization decorator.
`merge_dicts`(*dicts[, deep, list_strategy])	Merge multiple dictionaries.
`suggest_field_name`(invalid_name)	Suggest a valid field name from invalid one.
`unflatten_dict`(data[, separator])	Unflatten a dictionary.
`validate_field_name`(name)	Validate that a field name is valid Python identifier.

Module Contents¶

haive.core.schema.compatibility.utils.calculate_similarity(str1, str2)[source]¶

Calculate similarity between two strings (0-1).

Uses sequence matcher for basic similarity.

Parameters:

str1 (str)
str2 (str)

Return type:

float

haive.core.schema.compatibility.utils.create_example_value(type_hint)[source]¶

Create an example value for a type hint.

Parameters:: type_hint (type)
Return type:: Any

haive.core.schema.compatibility.utils.create_schema_diff(schema1, schema2)[source]¶

Create a diff between two schemas.

Returns dict with: - added_fields: Fields in schema2 but not schema1 - removed_fields: Fields in schema1 but not schema2 - modified_fields: Fields with different types/properties - unchanged_fields: Fields that are the same

Parameters:

schema1 (haive.core.schema.compatibility.types.SchemaInfo)
schema2 (haive.core.schema.compatibility.types.SchemaInfo)

Return type:

dict[str, Any]

haive.core.schema.compatibility.utils.estimate_memory_usage(schema)[source]¶

Estimate memory usage of a schema instance in bytes.

Parameters:: schema (haive.core.schema.compatibility.types.SchemaInfo)
Return type:: int

haive.core.schema.compatibility.utils.extract_path_value(data, path, separator='.', default=None)[source]¶

Extract value from nested dict using path.

Supports: - Dot notation: “user.profile.name” - Array indices: “items[0].name” - Wildcards: “items[*].name”

Parameters:

data (dict[str, Any])
path (str)
separator (str)
default (Any)

Return type:

Any

haive.core.schema.compatibility.utils.extract_type_name(type_hint)[source]¶

Extract a readable name from a type hint.

Parameters:: type_hint (type)
Return type:: str

haive.core.schema.compatibility.utils.find_similar_fields(target_field, source_fields, threshold=0.6)[source]¶

Find similar field names with scores.

Returns list of (field_name, similarity_score) tuples.

Parameters:

target_field (str)
source_fields (list[str])
threshold (float)

Return type:

list[tuple[str, float]]

haive.core.schema.compatibility.utils.flatten_nested_dict(data, parent_key='', separator='.')[source]¶

Flatten nested dictionary.

Examples

{“user”: {“name”: “John”, “age”: 30}} becomes {“user.name”: “John”, “user.age”: 30}

Parameters:

data (dict[str, Any])
parent_key (str)
separator (str)

Return type:

dict[str, Any]

haive.core.schema.compatibility.utils.format_type_path(types)[source]¶

Format a type conversion path for display.

Parameters:: types (list[type])
Return type:: str

haive.core.schema.compatibility.utils.generate_schema_hash(schema)[source]¶

Generate a hash for schema comparison.

Parameters:: schema (type[pydantic.BaseModel] | haive.core.schema.compatibility.types.SchemaInfo)
Return type:: str

haive.core.schema.compatibility.utils.get_all_subclasses(cls)[source]¶

Get all subclasses of a class recursively.

Parameters:: cls (type)
Return type:: set[type]

haive.core.schema.compatibility.utils.memoize(func)[source]¶

Simple memoization decorator.

Parameters:: func (collections.abc.Callable[Ellipsis, T])
Return type:: collections.abc.Callable[Ellipsis, T]

haive.core.schema.compatibility.utils.merge_dicts(*dicts, deep=True, list_strategy='extend')[source]¶

Merge multiple dictionaries.

Parameters:

*dicts (dict[str, Any]) – Dictionaries to merge
deep (bool) – Whether to deep merge nested dicts
list_strategy (str) – How to handle lists (“extend”, “replace”, “unique”)

Return type:

dict[str, Any]

haive.core.schema.compatibility.utils.suggest_field_name(invalid_name)[source]¶

Suggest a valid field name from invalid one.

Parameters:: invalid_name (str)
Return type:: str

haive.core.schema.compatibility.utils.unflatten_dict(data, separator='.')[source]¶

Unflatten a dictionary.

Examples

{“user.name”: “John”, “user.age”: 30} becomes {“user”: {“name”: “John”, “age”: 30}}

Parameters:

data (dict[str, Any])
separator (str)

Return type:

dict[str, Any]

haive.core.schema.compatibility.utils.validate_field_name(name)[source]¶

Validate that a field name is valid Python identifier.

Parameters:: name (str)
Return type:: bool