qat.experimental.system_data.materialisers.plugins module

Shared contracts and registry tooling for source-specific materialiser plugins.

Plugin authors should define two things in their source package:

  1. A SourceAdditionalDataModel subclass describing any extra inputs needed in

    addition to source_payload.

  2. A class implementing SourceMaterialiserPlugin with:

    • source_type and source_version identifiers,

    • resolve_type_and_version detector for payload-based source resolution,

    • additional_data_model pointing to the schema class,

    • verify_integrity for source trust/integrity checks,

    • materialise to build and return CanonicalSystemData.

Plugins are registered by calling register_materialiser_plugin from this module, typically as an import-side effect in the plugin module itself.

class SourceAdditionalDataModel(**data)

Bases: BaseModel

Base schema for source-specific additional-data payloads.

Subclass this model in each source package to define the typed source_additional_data contract expected by that plugin.

Design notes: - extra='forbid' rejects unknown keys at the boundary. - arbitrary_types_allowed=True permits rich compiler-owned types (for

example TargetData) in plugin models.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class SourceMaterialiserPlugin(*args, **kwargs)

Bases: Protocol

Structural interface for source/version materialiser plugins.

Minimal implementation pattern:

class MySourceAdditionalData(SourceAdditionalDataModel):
    helper: MyHelperType

class MySourcePlugin:
    source_type = SourceType.MY_SOURCE
    source_version = "1.0.0"
    additional_data_model = MySourceAdditionalData

    def resolve_type_and_version(
        self,
        source_payload: dict[str, Any],
    ) -> tuple[SourceType, str] | None:
        ...

    def verify_integrity(self, source_payload: dict[str, Any]) -> None:
        ...

    def materialise(
        self,
        *,
        source_payload: dict[str, Any],
        source_version: str,
        additional_data: MySourceAdditionalData,
    ) -> CanonicalSystemData:
        ...

Register the plugin via register_materialiser_plugin in this module.

additional_data_model: type[SourceAdditionalDataModel]
materialise(*, source_payload, source_version, additional_data)

Materialise canonical data for this source/version plugin.

additional_data is already validated by additional_data_model.model_validate before this method is called.

Return type:

CanonicalSystemData

resolve_type_and_version(source_payload)

Return this plugin descriptor when payload matches, else None.

The detector may use payload metadata and/or structural pattern checks, but it should stay lightweight and side-effect free. Avoid full payload validation or expensive normalisation in this method.

Return type:

tuple[SourceType, str] | None

source_type: SourceType
source_version: str
verify_integrity(source_payload)

Verify source payload integrity for this source/version.

Raise a MaterialisationError subclass for structured failures. Any unexpected exception may be wrapped by the boundary as a SourceIntegrityError.

Return type:

None

get_materialiser_plugin(*, source_type, source_version)

Return the registered plugin for source_type/source_version.

Return type:

SourceMaterialiserPlugin | None

get_registered_materialiser_plugins(source_type=None)

Return registered plugins in deterministic registry-key order.

When source_type is provided, only plugins for that source are returned.

Return type:

tuple[SourceMaterialiserPlugin, ...]

get_registered_source_versions(source_type)

Return sorted registered versions for a source type.

Return type:

tuple[str, ...]

register_materialiser_plugin(*, plugin, replace=False)

Register a materialiser plugin for source type/version dispatch.

Return type:

None