dataset_default

dataset_default

a class based approach for interacting with Domo Datasets

DomoDataset_Default dataclass

DomoDataset_Default(
    auth: DomoAuth,
    id: str,
    raw: dict,
    display_type: str = "",
    data_provider_type: str = "",
    name: str = "",
    description: str = "",
    row_count: int | None = None,
    column_count: int | None = None,
    stream_id: int | None = None,
    cloud_id: str | None = None,
    last_touched_dt: datetime | None = None,
    last_updated_dt: datetime | None = None,
    created_dt: datetime | None = None,
    owner: dict = dict(),
    formulas: dict = dict(),
    Data: DomoDataset_Data | None = None,
    Schema: DomoDataset_Schema | None = None,
    Stream: DomoStream | None = None,
    Tags: DomoTags | None = None,
    PDP: DatasetPdpPolicies | None = None,
    Access: DomoDatasetAccessController | None = None,
    Formulas: DomoDataset_FormulasManager | None = None,
    Certification: DomoCertification | None = None,
    AI_Readiness: DomoDataset_AI_Readiness | None = None,
    _is_schedule_hydrated: bool = False,
)

Bases: DomoEntity_w_Lineage

interacts with domo datasets

is_federated property

is_federated: bool

Heuristic: decide if a dataset JSON represents a federated (proxy) dataset.

is_manual property

is_manual: bool

True when composed schedule resolves to manual execution.

This property is strict by design: callers must either hydrate schedule state first or use async helpers that auto-hydrate.

is_schedule_hydrated property

is_schedule_hydrated: bool

Whether schedule state has been explicitly hydrated from stream metadata.

is_scheduled property

is_scheduled: bool

True when composed schedule resolves to scheduled execution.

is_view property

is_view: bool

Check if this dataset is a view.

transport_type property

transport_type: str | None

Get the transport type of the dataset if available.

Returns None if transport type cannot be determined rather than raising an exception. This allows the property to be safely included in serialization.

duplicate async

duplicate(
    new_name: str | None = None,
    *,
    new_description: str | None = None,
    override_data_provider_key: str | None = None,
    override_transport_description: str | None = None,
    override_configuration: list[dict] | None = None,
    override_account_id: int | None = None,
    is_execute: bool = False,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoDataset_Default

Duplicate this dataset by creating a new stream+dataset with the same connector config.

Creates a new dataset with the same connector type and configuration as this dataset. Optionally overrides the connector at creation time (useful for creating a dataset with a different connector from the start).

Parameters:

Name	Type	Description	Default
new_name	str | None	Name for the new dataset. Defaults to "{current_name} (Copy)".	None
new_description	str | None	Description for the new dataset. Defaults to current description.	None
override_data_provider_key	str | None	Override the connector type for the new dataset. Defaults to current stream's dataProvider.key.	None
override_transport_description	str | None	Override the transport description. Defaults to current stream's transport.description.	None
override_configuration	list[dict] | None	Override the stream configuration. Defaults to current stream's configuration.	None
override_account_id	int | None	Account ID for connectors requiring auth.	None
is_execute	bool	If True, run the stream immediately after creation.	False
context	RouteContext | None	Optional RouteContext for API call configuration.	None
**context_kwargs		Additional context parameters.	{}

Returns:

Name	Type	Description
DomoDataset_Default	DomoDataset_Default	The newly created dataset (hydrated).

Raises:

Type	Description
ValueError	If this dataset has no stream (cannot duplicate non-connector datasets this way).
Stream_CRUD_Error	If stream creation fails.

Example

ds = await DomoDataset.get_by_id(auth, dataset_id="abc123")

Simple duplicate with same connector

new_ds = await ds.duplicate()

Duplicate with a different connector

new_ds = await ds.duplicate( ... new_name="My Dataset (DomoStats)", ... override_data_provider_key="domostats", ... override_transport_description="com.domo.connector.domostats", ... override_configuration=[ ... {"name": "report", "value": "cards", "type": "string"}, ... ], ... is_execute=True, ... )

Note

Some built-in connectors (like domostats) may reuse existing streams when creating via POST /api/data/v1/streams. If you need a guaranteed new stream, use a connector that creates fresh streams (e.g. governance, third-party connectors).

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

async def duplicate(
    self,
    new_name: str | None = None,
    *,
    new_description: str | None = None,
    override_data_provider_key: str | None = None,
    override_transport_description: str | None = None,
    override_configuration: list[dict] | None = None,
    override_account_id: int | None = None,
    is_execute: bool = False,
    context: RouteContext | None = None,
    **context_kwargs,
) -> "DomoDataset_Default":
    """Duplicate this dataset by creating a new stream+dataset with the same connector config.

    Creates a new dataset with the same connector type and configuration as
    this dataset. Optionally overrides the connector at creation time (useful
    for creating a dataset with a different connector from the start).

    Args:
        new_name: Name for the new dataset. Defaults to
            "{current_name} (Copy)".
        new_description: Description for the new dataset. Defaults to
            current description.
        override_data_provider_key: Override the connector type for the
            new dataset. Defaults to current stream's dataProvider.key.
        override_transport_description: Override the transport description.
            Defaults to current stream's transport.description.
        override_configuration: Override the stream configuration.
            Defaults to current stream's configuration.
        override_account_id: Account ID for connectors requiring auth.
        is_execute: If True, run the stream immediately after creation.
        context: Optional RouteContext for API call configuration.
        **context_kwargs: Additional context parameters.

    Returns:
        DomoDataset_Default: The newly created dataset (hydrated).

    Raises:
        ValueError: If this dataset has no stream (cannot duplicate
            non-connector datasets this way).
        Stream_CRUD_Error: If stream creation fails.

    Example:
        >>> ds = await DomoDataset.get_by_id(auth, dataset_id="abc123")
        >>> # Simple duplicate with same connector
        >>> new_ds = await ds.duplicate()
        >>>
        >>> # Duplicate with a different connector
        >>> new_ds = await ds.duplicate(
        ...     new_name="My Dataset (DomoStats)",
        ...     override_data_provider_key="domostats",
        ...     override_transport_description="com.domo.connector.domostats",
        ...     override_configuration=[
        ...         {"name": "report", "value": "cards", "type": "string"},
        ...     ],
        ...     is_execute=True,
        ... )

    Note:
        Some built-in connectors (like domostats) may reuse existing
        streams when creating via POST /api/data/v1/streams. If you need
        a guaranteed new stream, use a connector that creates fresh streams
        (e.g. governance, third-party connectors).
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    # Hydrate stream if needed
    if not self.Stream or not self.Stream.id:
        await self.hydrate_schedule_state(force_refresh=True, context=context)

    if not self.Stream or not self.Stream.id:
        raise ValueError(
            f"Cannot duplicate dataset '{self.name}' ({self.id}): "
            "no stream found. Only connector-backed datasets can be duplicated."
        )

    # Resolve overrides
    dp_key = override_data_provider_key or self.Stream.data_provider_key
    transport_desc = (
        override_transport_description or self.Stream.transport_description
    )
    config = override_configuration
    if config is None:
        raw_config = (
            self.Stream.raw.get("configuration", []) if self.Stream.raw else []
        )
        # Internal config keys that cause 500 errors on create
        _skip_config_keys = {
            "retry.retryNumber",
            "updatemode.mode",
            "_description_",
        }
        config = [
            {
                "name": c.get("name"),
                "value": c.get("value"),
                "type": c.get("type", "string"),
                **({"category": c["category"]} if "category" in c else {}),
            }
            for c in raw_config
            if c.get("name") and c.get("name") not in _skip_config_keys
        ]

    ds_name = new_name or f"{self.name} (Copy)"
    ds_desc = new_description or self.description or ""

    # Build the create body
    # Note: API expects version "0" on create, not the source's version
    body: dict[str, Any] = {
        "transport": {
            "type": "CONNECTOR",
            "description": transport_desc,
            "version": "0",
        },
        "configuration": config,
        "updateMethod": self.Stream.update_method or "REPLACE",
        "dataProvider": {"key": dp_key},
        "dataSource": {
            "name": ds_name,
            "description": ds_desc,
        },
    }

    if override_account_id is not None:
        body["account"] = {"id": override_account_id}

    # Create the new stream+dataset
    from ...routes.dataset.stream import Stream_CRUD_Error, create_stream

    res = await create_stream(auth=self.auth, body=body, context=context)

    if not res.is_success:
        raise Stream_CRUD_Error(
            operation="create",
            message=f"Failed to duplicate dataset '{self.name}'",
            res=res,
        )

    # Extract the new dataset ID from the response
    new_dataset_id = res.response.get("dataSource", {}).get("id")
    if not new_dataset_id:
        raise Stream_CRUD_Error(
            operation="create",
            message="Stream creation succeeded but no dataset ID in response",
            res=res,
        )

    # Get the new dataset (hydrated)
    new_ds = await self.get_by_id(
        auth=self.auth,
        dataset_id=new_dataset_id,
        is_get_account=False,
        context=context,
    )

    # Optionally execute the stream
    if is_execute and new_ds.stream_id:
        from ...routes.dataset.stream import execute_stream

        await execute_stream(
            auth=self.auth,
            stream_id=str(new_ds.stream_id),
            context=context,
        )

    return new_ds

get_by_id async classmethod

get_by_id(
    auth: DomoAuth,
    dataset_id: str,
    debug_api: bool = False,
    return_raw: Literal[True] = True,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    is_use_default_dataset_class: bool = False,
    parent_class: str | None = None,
    is_get_account: bool = True,
    is_suppress_no_account_config: bool = True,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    max_subscriptions_to_check: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

get_by_id(
    auth: DomoAuth,
    dataset_id: str,
    debug_api: bool = False,
    return_raw: Literal[False] = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    is_use_default_dataset_class: bool = False,
    parent_class: str | None = None,
    is_get_account: bool = True,
    is_suppress_no_account_config: bool = True,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    max_subscriptions_to_check: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoDataset_Default

get_by_id(
    auth: DomoAuth,
    dataset_id: str,
    debug_api: bool = False,
    return_raw: bool = False,
    session: AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    is_use_default_dataset_class: bool = False,
    parent_class: str | None = None,
    is_get_account: bool = True,
    is_suppress_no_account_config: bool = True,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    max_subscriptions_to_check: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

retrieves dataset metadata

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
dataset_id	str	Dataset identifier to fetch	required
debug_api	bool	Enable verbose API logging	False
return_raw	bool	If True return the raw response object	False
check_if_published	bool | None	When True attempt to resolve publish subscriptions. If None and either parent_auth or parent_auth_retrieval_fn is provided, defaults to True.	None
parent_auth_retrieval_fn	Callable | None	Callable used to obtain publisher auth for publish checks	None
parent_auth	DomoAuth | None	Pre-existing publisher auth (alternative to parent_auth_retrieval_fn)	None
max_subscriptions_to_check	int | None	Optional limit passed to the PublishResolver	None

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

@classmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    dataset_id: str,
    debug_api: bool = False,
    return_raw: bool = False,
    session: httpx.AsyncClient | None = None,
    debug_num_stacks_to_drop: int = 2,
    is_use_default_dataset_class: bool = False,
    parent_class: str | None = None,
    is_get_account: bool = True,
    is_suppress_no_account_config: bool = True,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    parent_auth: DomoAuth | None = None,
    max_subscriptions_to_check: int | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """retrieves dataset metadata

    Args:
        auth: Authentication object for API requests
        dataset_id: Dataset identifier to fetch
        debug_api: Enable verbose API logging
        return_raw: If True return the raw response object
        check_if_published: When True attempt to resolve publish subscriptions.
            If None and either parent_auth or parent_auth_retrieval_fn is provided, defaults to True.
        parent_auth_retrieval_fn: Callable used to obtain publisher auth for publish checks
        parent_auth: Pre-existing publisher auth (alternative to parent_auth_retrieval_fn)
        max_subscriptions_to_check: Optional limit passed to the PublishResolver
    """
    # Auto-enable publish checking when parent auth is provided (either directly or via retrieval fn)
    # unless explicitly disabled
    if (parent_auth_retrieval_fn or parent_auth) and check_if_published is None:
        check_if_published = True

    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        **context_kwargs,
    )
    # Override parent_class if explicitly provided
    if parent_class:
        context.parent_class = parent_class

    res = await dataset_routes.get_dataset_by_id(
        auth=auth,
        dataset_id=dataset_id,
        context=context,
    )

    if return_raw:
        return res

    obj = res.response

    # Validate response is a dict
    if not isinstance(obj, dict):
        raise ValueError(
            f"Expected dict response from API, got {type(obj).__name__}: {obj}"
        )

    # Check if published (for federated datasets)
    # Only attempt probe if we have a way to get parent auth
    is_published = False
    subscription = None
    if (
        check_if_published
        and cls._is_federated(obj)
        and (parent_auth_retrieval_fn or parent_auth)
    ):
        # TODO: Future enhancement - use lineage-based class selection
        # Instead of checking here, we could:
        # 1. Create dataset with from_dict
        # 2. Call: lineage = await DomoLineage.get_lineage_from_entity(
        #       entity=ds, check_is_published=True, parent_auth_retrieval_fn=...)
        # 3. Use FederationContext on the returned DomoDataset for published/federated behavior
        # For now, keep existing probe pattern
        await logger.info(
            f"Lineage publish probe: Testing for published state on federated entity {dataset_id} "
            f"with parent_auth={'set' if parent_auth else 'unset'}, "
            f"parent_auth_retrieval_fn={'set' if parent_auth_retrieval_fn else 'unset'}"
        )
        await logger.debug(
            f"Checking if federated dataset {dataset_id} is published",
            extra={
                "dataset_id": dataset_id,
                "is_federated": True,
                "check_if_published": check_if_published,
            },
        )
        probe = await cls.probe_is_published(
            entity_id=dataset_id,
            subscriber_auth=auth,
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            parent_auth=parent_auth,
            session=session,
            debug_api=debug_api,
            max_subscriptions_to_check=max_subscriptions_to_check,
            context=context,
        )
        is_published = probe.is_published
        subscription = probe.subscription

        await logger.debug(
            f"Probe results: is_published={is_published}, has_subscription={subscription is not None}",
            extra={
                "dataset_id": dataset_id,
                "is_published": is_published,
                "has_subscription": subscription is not None,
                "subscription_id": subscription.id if subscription else None,
            },
        )

    ds = cls.from_dict(
        obj=obj,
        auth=auth,
        new_cls=cls,
        is_use_default_dataset_class=is_use_default_dataset_class,
        check_is_published=is_published,  # Used for class selection only
        parent_auth_retrieval_fn=parent_auth_retrieval_fn,
        parent_auth=parent_auth,
    )

    if is_published and subscription:
        await logger.debug(
            f"Hydrating publish helper for dataset {dataset_id}",
            extra={
                "dataset_id": dataset_id,
                "subscription_id": subscription.id,
            },
        )
        helper = ds.enable_federation_support()
        helper.hydrate_from_existing(
            subscription=subscription,
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            content_type="DATA_SOURCE",
            entity_id=str(dataset_id),
        )
        # Update Lineage.Federation so publish info is accessible via ds.Lineage.Federation
        if ds.Lineage:
            ds.Lineage.Federation = helper
            await logger.info(
                f"✅ Lineage.Federation assigned for dataset {dataset_id}",
                extra={
                    "dataset_id": dataset_id,
                    "subscription_id": subscription.id,
                },
            )
        else:
            await logger.warning(
                f"⚠️ Lineage is None, cannot assign Federation for dataset {dataset_id}",
                extra={"dataset_id": dataset_id},
            )
    elif check_if_published and cls._is_federated(obj):
        await logger.debug(
            f"Skipping publish helper hydration: is_published={is_published}, has_subscription={subscription is not None}",
            extra={
                "dataset_id": dataset_id,
                "is_published": is_published,
                "has_subscription": subscription is not None,
            },
        )

    if ds.Stream:
        await ds.Stream.refresh(
            is_get_account=is_get_account,
            is_suppress_no_account_config=is_suppress_no_account_config,
            context=context,
        )

    # Auto-trace lineage if parent_auth_retrieval_fn is provided
    if parent_auth_retrieval_fn and ds.Lineage:
        await ds.Lineage.get(
            parent_auth_retrieval_fn=parent_auth_retrieval_fn,
            parent_auth=parent_auth,
            session=session,
            debug_api=debug_api,
            context=context,
        )

    return ds

get_is_manual async

get_is_manual() -> bool

Get manual/scheduled state with guaranteed schedule hydration.

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

async def get_is_manual(self) -> bool:
    """Get manual/scheduled state with guaranteed schedule hydration."""
    await self.hydrate_schedule_state()
    return self.is_manual

get_is_scheduled async

get_is_scheduled() -> bool

Get scheduled state with guaranteed schedule hydration.

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

async def get_is_scheduled(self) -> bool:
    """Get scheduled state with guaranteed schedule hydration."""
    await self.hydrate_schedule_state()
    return self.is_scheduled

hydrate_schedule_state async

hydrate_schedule_state(
    force_refresh: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoDataset_Default

Hydrate schedule state for this dataset.

Search-index dataset payloads omit streamId and schedule fields, so this method rehydrates dataset metadata by ID and fetches stream metadata when available.

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

async def hydrate_schedule_state(
    self,
    force_refresh: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> "DomoDataset_Default":
    """Hydrate schedule state for this dataset.

    Search-index dataset payloads omit `streamId` and schedule fields, so this
    method rehydrates dataset metadata by ID and fetches stream metadata when
    available.
    """
    if self._is_schedule_hydrated and not force_refresh:
        return self

    context = RouteContext.build_context(context=context, **context_kwargs)

    # Ensure we have full dataset metadata (search index payloads can be sparse).
    if force_refresh or self.stream_id is None:
        refreshed = await self.get_by_id(
            auth=self.auth,
            dataset_id=self.id,
            is_get_account=False,
            context=context,
        )

        self.raw = refreshed.raw
        self.stream_id = refreshed.stream_id
        self.data_provider_type = refreshed.data_provider_type
        self.display_type = refreshed.display_type
        self.name = refreshed.name
        self.description = refreshed.description
        self.row_count = refreshed.row_count
        self.column_count = refreshed.column_count
        self.owner = refreshed.owner
        self.formulas = refreshed.formulas

    if self.stream_id is not None:
        self.Stream = await dmdst.DomoStream.get_by_id(
            auth=self.auth,
            stream_id=str(self.stream_id),
            is_get_account=False,
            context=context,
        )
    else:
        self.Stream = None

    self._is_schedule_hydrated = True
    return self

to_dict

to_dict(
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
) -> dict

Serialize dataset and composed objects for export/debugging.

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

def to_dict(
    self, override_fn: Callable | None = None, return_snake_case: bool = False
) -> dict:
    """Serialize dataset and composed objects for export/debugging."""
    if override_fn:
        return override_fn(self)

    result = super().to_dict(return_snake_case=return_snake_case)

    if self.Stream:
        stream_key = "stream" if return_snake_case else "Stream"
        result[stream_key] = self.Stream.to_dict(
            return_snake_case=return_snake_case
        )

    return result

update_properties async

update_properties(
    name: str | None = None,
    description: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs
)

Update this datasource's name and/or description.

Applies to datasets, dataset views, and data models alike. Updates the local name/description on success.

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

async def update_properties(
    self,
    name: str | None = None,
    description: str | None = None,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
):
    """Update this datasource's name and/or description.

    Applies to datasets, dataset views, and data models alike. Updates the
    local ``name``/``description`` on success.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    res = await dataset_routes.update_dataset_properties(
        auth=self.auth,
        dataset_id=self.id,
        name=name,
        description=description,
        context=context,
    )
    if name is not None:
        self.name = name
    if description is not None:
        self.description = description
    return res

DomoDataset_NoTransportType_Error

DomoDataset_NoTransportType_Error(
    cls_instance=None, message: str | None = None, **kwargs
)

Bases: ClassError

Raised when unable to determine the transport type of a dataset.

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

def __init__(self, cls_instance=None, message: str | None = None, **kwargs):
    if not message:
        message = "Unable to determine the transport type of the dataset."

    super().__init__(cls_instance=cls_instance, message=message, **kwargs)