manager_data_model

manager_data_model

Data-model definition manager for DomoSemanticModel.

Provides access to a data model's tables (objects) and JOIN criteria (relationships) via the composition pattern (model.Model.get(), model.Model.tables, model.Model.relationships, ...).

Mirrors DomoDatasetView_DefinitionManager, but parses the model key of the schema/indexed response into a DataModelTemplate rather than the dataset-view viewTemplate.

DomoSemanticModel_ModelManager dataclass

DomoSemanticModel_ModelManager(
    parent: DomoDataset_Default = None,
    template: DataModelTemplate | None = None,
)

Bases: DomoSubEntity

Manager for a data model's definition.

Provides typed access to: - DataModelTemplate with the model's objects (tables) and relationships (joins) - Convenience properties: tables, relationships, source_dataset_ids

Usage

await model.Model.get() model.Model.template # DataModelTemplate model.Model.tables # list[DataModelObject] model.Model.relationships # list[DataModelRelationship] model.Model.source_dataset_ids # set[str]

relationships property

relationships: list[DataModelRelationship]

JOIN criteria between the model's tables. Empty if not loaded.

source_dataset_ids property

source_dataset_ids: set[str]

Source dataset IDs this model depends on. Empty if not loaded.

tables property

tables: list[DataModelObject]

Source-dataset tables in this model. Empty if not loaded.

get async

get(
    *, context: RouteContext | None = None, **context_kwargs
) -> DataModelTemplate

Fetch the data-model definition for this semantic model.

Populates self.template with a typed DataModelTemplate.

Returns:

Type	Description
DataModelTemplate	DataModelTemplate instance with typed access to tables and joins.

Source code in src/crew_dcs/classes/DomoDataset/manager_data_model.py

async def get(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DataModelTemplate:
    """Fetch the data-model definition for this semantic model.

    Populates self.template with a typed DataModelTemplate.

    Returns:
        DataModelTemplate instance with typed access to tables and joins.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await dataset_routes.get_dataset_view_schema_indexed(
        auth=self.auth,
        dataset_id=self.parent.id,
        flatten=False,  # required to surface model.objects + modeledColumns
        context=context,
    )

    if res.is_success:
        self.template = DataModelTemplate.from_dict(res.response)

    return self.template

update async

update(
    data_source_name: str | None = None,
    data_source_description: str | None = None,
    last_updated: str | None = None,
    schema: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Persist the model's tables + JOIN criteria.

By default serializes the loaded template (after edits) via DataModelTemplate.to_schema() and PUTs it. Pass schema to send a pre-built {"objects": ..., "relationships": ...} payload instead.

Raises:

Type	Description
ValueError	if no schema is provided and the template hasn't been loaded (call await model.Model.get() first).

Source code in src/crew_dcs/classes/DomoDataset/manager_data_model.py

async def update(
    self,
    data_source_name: str | None = None,
    data_source_description: str | None = None,
    last_updated: str | None = None,
    schema: dict | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Persist the model's tables + JOIN criteria.

    By default serializes the loaded ``template`` (after edits) via
    ``DataModelTemplate.to_schema()`` and PUTs it. Pass ``schema`` to send
    a pre-built ``{"objects": ..., "relationships": ...}`` payload instead.

    Raises:
        ValueError: if no schema is provided and the template hasn't been
            loaded (call ``await model.Model.get()`` first).
    """
    if schema is None:
        if self.template is None:
            raise ValueError(
                "No definition loaded — call `await model.Model.get()` "
                "before update(), or pass schema=..."
            )
        schema = self.template.to_schema()

    context = RouteContext.build_context(context=context, **context_kwargs)
    return await dataset_routes.update_semantic_model(
        auth=self.auth,
        model_id=self.parent.id,
        schema=schema,
        data_source_name=data_source_name,
        data_source_description=data_source_description,
        last_updated=last_updated,
        context=context,
    )

validate_relationships async

validate_relationships(
    sample_size: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[JoinValidationResult]

Validate the model's JOIN keys by sampling real data.

For each relationship, samples sample_size values from each join-key column on both sides and checks:

• Overlap: do the sampled values from the left key appear in the right key? Low overlap signals a wrong column choice.
• Distinctness: are the values unique on the assumed "one" side? Duplicates on the dimension side signal a wrong cardinality direction.
• Cardinality hint: compares duplicate ratios to suggest which side is the dimension (one) and which is the fact (many).

Requires that the template has been loaded (call await model.Model.get() first).

Parameters:

Name	Type	Description	Default
sample_size	int	Number of rows to sample per join key (default 100).	100
context	RouteContext | None	Optional RouteContext.	None
**context_kwargs		Additional context parameters.	{}

Returns:

Type	Description
list[JoinValidationResult]	List of JoinValidationResult, one per relationship.

Raises:

Type	Description
ValueError	If the template hasn't been loaded.

Source code in src/crew_dcs/classes/DomoDataset/manager_data_model.py

async def validate_relationships(
    self,
    sample_size: int = 100,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[JoinValidationResult]:
    """Validate the model's JOIN keys by sampling real data.

    For each relationship, samples ``sample_size`` values from each
    join-key column on both sides and checks:

    - **Overlap**: do the sampled values from the left key appear in
      the right key? Low overlap signals a wrong column choice.
    - **Distinctness**: are the values unique on the assumed "one"
      side? Duplicates on the dimension side signal a wrong
      cardinality direction.
    - **Cardinality hint**: compares duplicate ratios to suggest
      which side is the dimension (one) and which is the fact (many).

    Requires that the template has been loaded (call
    ``await model.Model.get()`` first).

    Args:
        sample_size: Number of rows to sample per join key (default 100).
        context: Optional RouteContext.
        **context_kwargs: Additional context parameters.

    Returns:
        List of JoinValidationResult, one per relationship.

    Raises:
        ValueError: If the template hasn't been loaded.
    """
    if self.template is None:
        raise ValueError(
            "No definition loaded — call `await model.Model.get()`"
            " before validate_relationships()"
        )

    return await validate_relationships(
        auth=self.auth,
        template=self.template,
        sample_size=sample_size,
        context=context,
        **context_kwargs,
    )