data_model_converter

data_model_converter

DataModelConverter — converts DataModelTemplate to OSI semantic models.

DataModelConverter

Bases: OsiConverter

Converts a Domo DataModelTemplate to a list of OsiSemanticModels.

One OsiSemanticModel is produced per DataModelObject (source table). Join key columns become OsiEntity instances; all other columns are classified as dimensions or metrics based on AI_Readiness subType, falling back to data-type inference.

Cross-model relationships (JOINs) are attached to the left-side model.

Metrics (Phase 1.5) come from two routes, per the PRD's classification rules — both can contribute to the same model, they are not mutually exclusive: 1. Primary: columns with AI_Readiness_Column.subType == 'METRIC'. 2. Fallback: Beast Modes with BeastModeTemplate.aggregated is True, passed in via the beastmodes param (keyed by datasource_id). A Beast Mode whose name collides with an AI-Readiness-derived metric name is skipped — the AI Readiness path wins.

Usage

converter = DataModelConverter() models = converter.convert(template, datasets) converter.export_to_file(models, "output.yaml")

With Beast Mode metrics:

models = converter.convert( template, datasets, beastmodes={"ds-orders": [aggregated_template, ...]}, domo_instance="mycompany", )

convert

convert(
    source,
    datasets: dict,
    beastmodes: (
        dict[str, list[BeastModeTemplate]] | None
    ) = None,
    domo_instance: str | None = None,
    **options
) -> list[OsiSemanticModel]

Convert a DataModelTemplate to OSI semantic models.

Parameters:

Name	Type	Description	Default
source		DataModelTemplate with objects and relationships	required
datasets	dict	dict mapping datasource_id -> DomoDataset (or None)	required
beastmodes	dict[str, list[BeastModeTemplate]] | None	dict mapping datasource_id -> list[BeastModeTemplate] (Phase 1.5). Only templates with aggregated is True are converted to metrics; refs must already be resolved via BeastModeTemplate.resolve_domo_beast_mode_refs() or conversion raises UnresolvedBeastModeReferenceError.	None
domo_instance	str | None	Domo instance hostname, recorded on Beast-Mode- derived metrics' custom_extensions block. Optional — omitted from output when not provided.	None
**options		Unused beyond the params above	{}

Returns:

Type	Description
list[OsiSemanticModel]	list[OsiSemanticModel], one per DataModelObject

Source code in src/crew_dcs/integrations/osi/data_model_converter.py

def convert(  # noqa: C901
    self,
    source,
    datasets: dict,
    beastmodes: dict[str, list[BeastModeTemplate]] | None = None,
    domo_instance: str | None = None,
    **options,
) -> list[OsiSemanticModel]:
    """Convert a DataModelTemplate to OSI semantic models.

    Args:
        source: DataModelTemplate with objects and relationships
        datasets: dict mapping datasource_id -> DomoDataset (or None)
        beastmodes: dict mapping datasource_id -> list[BeastModeTemplate]
            (Phase 1.5). Only templates with ``aggregated is True`` are
            converted to metrics; refs must already be resolved via
            ``BeastModeTemplate.resolve_domo_beast_mode_refs()`` or
            conversion raises ``UnresolvedBeastModeReferenceError``.
        domo_instance: Domo instance hostname, recorded on Beast-Mode-
            derived metrics' ``custom_extensions`` block. Optional —
            omitted from output when not provided.
        **options: Unused beyond the params above

    Returns:
        list[OsiSemanticModel], one per DataModelObject
    """
    beastmodes = beastmodes or {}
    models: dict[str, OsiSemanticModel] = {}

    # Collect all join keys per object alias
    all_join_keys: dict[str, set[str]] = {}
    for rel in source.relationships:
        all_join_keys.setdefault(rel.left, set()).update(rel.left_keys or [])
        all_join_keys.setdefault(rel.right, set()).update(rel.right_keys or [])

    # Build one OsiSemanticModel per DataModelObject
    for alias, obj in source.objects.items():
        dataset = datasets.get(obj.datasource_id)
        ar_map = self._get_ai_readiness_map(dataset)
        join_keys = all_join_keys.get(alias, set())

        entities = []
        dimensions = []
        metrics = []

        for col in obj.columns or []:
            col_name = col.name
            col_type = col.type if hasattr(col, "type") else ""
            classification = self._classify_column(
                col_name, col_type, join_keys, ar_map
            )

            if classification == "entity":
                # Determine primary vs foreign: primary if this alias appears
                # on the left side of any relationship using this key
                is_primary = any(
                    rel.left == alias and col_name in (rel.left_keys or [])
                    for rel in source.relationships
                )
                entities.append(self._build_entity(col_name, is_primary=is_primary))

            elif classification in ("dimension", "time_dimension"):
                dimensions.append(
                    self._build_dimension(col_name, col_type, ar_map.get(col_name))
                )
            elif classification == "metric":
                metrics.append(
                    self._build_metric_from_ai_readiness(
                        col_name, col_type, ar_map.get(col_name)
                    )
                )

        # Beast Mode fallback (Phase 1.5): aggregated Beast Modes linked
        # to this object's dataset. A name collision with an
        # AI-Readiness-derived metric defers to the primary path.
        existing_metric_names = {m.name for m in metrics}
        for template in beastmodes.get(obj.datasource_id, []):
            if not template.aggregated or template.name in existing_metric_names:
                continue
            metrics.append(
                self._build_metric_from_beastmode(
                    template, obj.datasource_id, domo_instance
                )
            )
            existing_metric_names.add(template.name)

        name = (dataset.name if dataset else None) or alias
        description = ""
        if dataset and dataset.AI_Readiness:
            description = dataset.AI_Readiness.unit_of_analysis or ""

        domo_ext: dict = {"dataset_id": obj.datasource_id}
        if dataset and dataset.AI_Readiness and dataset.AI_Readiness.columns:
            domo_ext["agent_enabled"] = any(
                c.agentEnabled for c in dataset.AI_Readiness.columns
            )

        models[alias] = OsiSemanticModel(
            name=name,
            source_dataset_id=obj.datasource_id,
            description=description,
            entities=entities,
            dimensions=dimensions,
            metrics=metrics,
            domo_extensions=domo_ext,
        )

    # Attach cross-model relationships to the left-side model
    for rel in source.relationships:
        for left_key, right_key in zip(rel.left_keys or [], rel.right_keys or []):  # noqa: B905
            left_model = models.get(rel.left)
            right_model = models.get(rel.right)
            relationship = OsiRelationship(
                from_model=left_model.name if left_model else rel.left,
                to_model=right_model.name if right_model else rel.right,
                from_field=left_key,
                to_field=right_key,
                join_type=rel.join_type or "INNER",
                cardinality=rel.cardinality,
            )
            if left_model is not None:
                left_model.relationships.append(relationship)

    return list(models.values())