haive.core.schema.compatibility.protocols

from typing import Any. Protocol definitions for extending the schema compatibility system.

Classes

AsyncConverter

Protocol for async converters.

CompatibilityPlugin

Protocol for compatibility checker plugins.

ConversionStrategy

Protocol for conversion strategies.

ExampleFieldResolver

Example field resolver using similarity matching.

ExampleTypeInspector

Example type inspector for custom types.

FieldResolver

Protocol for resolving field mappings.

FieldTransformer

Protocol for field transformation functions.

PluginManager

Manages plugins for the compatibility system.

SchemaConvertible

Protocol for objects that can be converted to/from schemas.

SchemaEvolution

Protocol for schema evolution/migration.

SchemaRegistry

Protocol for schema registries.

SchemaValidator

Protocol for schema validators.

TypeInspector

Protocol for custom type inspection.

Functions

compatibility_plugin([priority])

Decorator to register a compatibility plugin.

converter_plugin(cls)

Decorator to register a converter plugin.

validator_plugin(cls)

Decorator to register a validator plugin.

Module Contents

class haive.core.schema.compatibility.protocols.AsyncConverter[source]

Bases: Protocol

Protocol for async converters.

async aconvert(value, context)[source]

Async conversion.

Parameters:
Return type:

Any

property supports_sync: bool

Whether sync conversion is also supported.

Return type:

bool

class haive.core.schema.compatibility.protocols.CompatibilityPlugin[source]

Bases: Protocol

Protocol for compatibility checker plugins.

check_compatibility(source_type, target_type)[source]

Check compatibility between types.

Parameters:
Return type:

haive.core.schema.compatibility.types.CompatibilityLevel | None

enhance_report(report, source, target)[source]

Enhance compatibility report with additional info.

Parameters:
Return type:

None

property name: str

Plugin name.

Return type:

str

property priority: int

Plugin priority (higher = runs first).

Return type:

int

class haive.core.schema.compatibility.protocols.ConversionStrategy[source]

Bases: Protocol

Protocol for conversion strategies.

can_convert(source, target)[source]

Check if strategy can handle conversion.

Parameters:
Return type:

bool

convert(value, source_type, target_type, context)[source]

Perform conversion.

Parameters:
Return type:

Any

property name: str

Strategy name.

Return type:

str

class haive.core.schema.compatibility.protocols.ExampleFieldResolver[source]

Example field resolver using similarity matching.

resolve_field(source_fields, target_field)[source]

Resolve by name similarity.

Parameters:
Return type:

str | None

suggest_mapping(source_schema, target_schema)[source]

Suggest mappings for all fields.

Parameters:
Return type:

dict[str, str]

class haive.core.schema.compatibility.protocols.ExampleTypeInspector[source]

Example type inspector for custom types.

can_inspect(type_hint)[source]

Check if type has custom metadata.

Parameters:

type_hint (type)

Return type:

bool

extract_constraints(type_hint)[source]

Extract validation constraints.

Parameters:

type_hint (type)

Return type:

dict[str, Any]

inspect(type_hint)[source]

Extract custom metadata.

Parameters:

type_hint (type)

Return type:

dict[str, Any]

class haive.core.schema.compatibility.protocols.FieldResolver[source]

Bases: Protocol

Protocol for resolving field mappings.

resolve_field(source_fields, target_field)[source]

Resolve source field for a target field.

Parameters:
Return type:

str | None

suggest_mapping(source_schema, target_schema)[source]

Suggest field mappings.

Parameters:
Return type:

dict[str, str]

class haive.core.schema.compatibility.protocols.FieldTransformer[source]

Bases: Protocol

Protocol for field transformation functions.

class haive.core.schema.compatibility.protocols.PluginManager[source]

Manages plugins for the compatibility system.

Init .

Returns:

Add return description]

Return type:

[TODO

get_compatibility_plugins()[source]

Get all registered compatibility plugins.

Return type:

list[CompatibilityPlugin]

get_converters()[source]

Get all registered converters.

Return type:

list[ConversionStrategy]

get_inspectors()[source]

Get all registered inspectors.

Return type:

list[TypeInspector]

get_resolvers()[source]

Get all registered resolvers.

Return type:

list[FieldResolver]

get_validators()[source]

Get all registered validators.

Return type:

list[SchemaValidator]

register_compatibility_plugin(plugin)[source]

Register a compatibility plugin.

Parameters:

plugin (CompatibilityPlugin)

Return type:

None

register_converter(converter)[source]

Register a conversion strategy.

Parameters:

converter (ConversionStrategy)

Return type:

None

register_inspector(inspector)[source]

Register a type inspector.

Parameters:

inspector (TypeInspector)

Return type:

None

register_resolver(resolver)[source]

Register a field resolver.

Parameters:

resolver (FieldResolver)

Return type:

None

register_validator(validator)[source]

Register a schema validator.

Parameters:

validator (SchemaValidator)

Return type:

None

class haive.core.schema.compatibility.protocols.SchemaConvertible[source]

Bases: Protocol

Protocol for objects that can be converted to/from schemas.

classmethod from_schema(schema)[source]

Create instance from a Pydantic schema.

Parameters:

schema (type[pydantic.BaseModel])

Return type:

T

to_schema()[source]

Convert to a Pydantic schema.

Return type:

type[pydantic.BaseModel]

class haive.core.schema.compatibility.protocols.SchemaEvolution[source]

Bases: Protocol

Protocol for schema evolution/migration.

can_migrate(from_version, to_version)[source]

Check if migration is possible.

Parameters:

  • from_version (str)

  • to_version (str)

Return type:

bool

migrate(data, from_version, to_version)[source]

Migrate data between schema versions.

Parameters:
Return type:

dict[str, Any]

property version: str

Schema version.

Return type:

str

class haive.core.schema.compatibility.protocols.SchemaRegistry[source]

Bases: Protocol

Protocol for schema registries.

find_compatible(target, min_score=0.7)[source]

Find compatible schemas with scores.

Parameters:

  • target (type[pydantic.BaseModel])

  • min_score (float)

Return type:

list[tuple[str, float]]

get(name)[source]

Get a schema by name.

Parameters:

name (str)

Return type:

type[pydantic.BaseModel] | None

list_schemas()[source]

List all registered schema names.

Return type:

list[str]

register(name, schema)[source]

Register a schema.

Parameters:

  • name (str)

  • schema (type[pydantic.BaseModel])

Return type:

None

class haive.core.schema.compatibility.protocols.SchemaValidator[source]

Bases: Protocol

Protocol for schema validators.

validate_compatibility(source, target)[source]

Validate compatibility between schemas.

Parameters:
Return type:

list[str]

validate_schema(schema)[source]

Validate a schema and return list of issues.

Parameters:

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

Return type:

list[str]

class haive.core.schema.compatibility.protocols.TypeInspector[source]

Bases: Protocol

Protocol for custom type inspection.

can_inspect(type_hint)[source]

Check if this inspector can handle the type.

Parameters:

type_hint (type)

Return type:

bool

extract_constraints(type_hint)[source]

Extract validation constraints from type.

Parameters:

type_hint (type)

Return type:

dict[str, Any]

inspect(type_hint)[source]

Inspect the type and return metadata.

Parameters:

type_hint (type)

Return type:

dict[str, Any]

haive.core.schema.compatibility.protocols.compatibility_plugin(priority=0)[source]

Decorator to register a compatibility plugin.

Parameters:

priority (int)

haive.core.schema.compatibility.protocols.converter_plugin(cls)[source]

Decorator to register a converter plugin.

Return type:

Any

haive.core.schema.compatibility.protocols.validator_plugin(cls)[source]

Decorator to register a validator plugin.

Return type:

Any