haive.core.schema.compatibility.mergers¶

Schema merging strategies for combining multiple schemas.

Classes¶

`ConflictResolution`	How to resolve field conflicts during merge.
`IntersectionMergeStrategy`	Include only fields present in all schemas.
`MergeContext`	Context for merge operations.
`MergeStrategy`	Abstract base for merge strategies.
`SchemaMerger`	Main schema merging engine.
`UnionMergeStrategy`	Include all fields from all schemas.

Functions¶

`create_intersection_schema`(*schemas[, name])	Create an intersection of multiple schemas.
`create_union_schema`(*schemas[, name])	Create a union of multiple schemas.
`merge_schemas`(schemas[, strategy, name])	Merge multiple schemas using specified strategy.

Module Contents¶

class haive.core.schema.compatibility.mergers.ConflictResolution[source]¶

Bases: str, enum.Enum

How to resolve field conflicts during merge.

Initialize self. See help(type(self)) for accurate signature.

class haive.core.schema.compatibility.mergers.IntersectionMergeStrategy[source]¶

Bases: MergeStrategy

Include only fields present in all schemas.

merge_fields(field_infos, context)[source]¶

Merge field definitions.

Parameters:

field_infos (list[tuple[str, haive.core.schema.compatibility.types.FieldInfo]])
context (MergeContext)

Return type:

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

should_include_field(field_name, schemas_with_field, total_schemas)[source]¶

Include only if in all schemas.

Parameters:

field_name (str)
schemas_with_field (list[str])
total_schemas (int)

Return type:

bool

class haive.core.schema.compatibility.mergers.MergeContext[source]¶

Context for merge operations.

Init .

Returns:: Add return description]
Return type:: [TODO

add_conflict(field_name, schemas, reason, resolution)[source]¶

Log a conflict.

Parameters:

field_name (str)
schemas (list[str])
reason (str)
resolution (str)

Return type:

None

add_warning(warning)[source]¶

Add a warning.

Parameters:: warning (str)
Return type:: None

track_field_source(field_name, schema_name)[source]¶

Track which schema a field came from.

Parameters:

field_name (str)
schema_name (str)

Return type:

None

class haive.core.schema.compatibility.mergers.MergeStrategy[source]¶

Bases: abc.ABC

Abstract base for merge strategies.

abstractmethod merge_fields(field_infos, context)[source]¶

Merge multiple field definitions.

Parameters:

field_infos (list[tuple[str, haive.core.schema.compatibility.types.FieldInfo]])
context (MergeContext)

Return type:

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

abstractmethod should_include_field(field_name, schemas_with_field, total_schemas)[source]¶

Determine if field should be included in merged schema.

Parameters:

field_name (str)
schemas_with_field (list[str])
total_schemas (int)

Return type:

bool

class haive.core.schema.compatibility.mergers.SchemaMerger(strategy='union', analyzer=None, compatibility_checker=None)[source]¶

Main schema merging engine.

Init .

Parameters:

strategy (MergeStrategy | str) – [TODO: Add description]
analyzer (haive.core.schema.compatibility.analyzer.TypeAnalyzer | None) – [TODO: Add description]
compatibility_checker (haive.core.schema.compatibility.compatibility.CompatibilityChecker | None) – [TODO: Add description]

merge_schemas(schemas, name=None, base_class=None)[source]¶

Merge multiple schemas into one.

Parameters:

schemas (list[type[pydantic.BaseModel] | haive.core.schema.compatibility.types.SchemaInfo]) – List of schemas to merge
name (str | None) – Name for the merged schema
base_class (type[pydantic.BaseModel] | None) – Base class for the merged schema

Returns:

Merged schema class

Return type:

type[pydantic.BaseModel]

class haive.core.schema.compatibility.mergers.UnionMergeStrategy(conflict_resolution=ConflictResolution.LAST_WINS)[source]¶

Bases: MergeStrategy

Include all fields from all schemas.

Init .

Parameters:: conflict_resolution (ConflictResolution) – [TODO: Add description]

merge_fields(field_infos, context)[source]¶

Merge field definitions.

Parameters:

field_infos (list[tuple[str, haive.core.schema.compatibility.types.FieldInfo]])
context (MergeContext)

Return type:

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

should_include_field(field_name, schemas_with_field, total_schemas)[source]¶

Include all fields.

Parameters:

field_name (str)
schemas_with_field (list[str])
total_schemas (int)

Return type:

bool

haive.core.schema.compatibility.mergers.create_intersection_schema(*schemas, name=None)[source]¶

Create an intersection of multiple schemas.

Parameters:

schemas (type[pydantic.BaseModel] | haive.core.schema.compatibility.types.SchemaInfo)
name (str | None)

Return type:

type[pydantic.BaseModel]

haive.core.schema.compatibility.mergers.create_union_schema(*schemas, name=None)[source]¶

Create a union of multiple schemas.

Parameters:

schemas (type[pydantic.BaseModel] | haive.core.schema.compatibility.types.SchemaInfo)
name (str | None)

Return type:

type[pydantic.BaseModel]

haive.core.schema.compatibility.mergers.merge_schemas(schemas, strategy='union', name=None)[source]¶

Merge multiple schemas using specified strategy.

Parameters:

schemas (list[type[pydantic.BaseModel] | haive.core.schema.compatibility.types.SchemaInfo])
strategy (str)
name (str | None)

Return type:

type[pydantic.BaseModel]