manager

manager

DomoDatasets Manager Class for collection operations on datasets.

DomoDatasets dataclass

DomoDatasets(
    auth: DomoAuth, datasets: list[DomoDataset] = list()
)

Bases: DomoManager

Manager class for searching and retrieving multiple datasets.

Provides methods for listing and searching datasets in a Domo instance.

get async

get(
    search_text: str | None = None,
    maximum: int | None = None,
    data_provider_type: str | None = None,
    hydrate_schedule_state: bool = False,
    hydrate_concurrency: int = 10,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoDataset]

Retrieve datasets from Domo, optionally filtering by search text.

Parameters:

Name	Type	Description	Default
search_text	str | None	Optional text to search for in dataset names	None
maximum	int | None	Maximum number of datasets to return	None
data_provider_type	str | None	Optional dataset provider type to filter on	None
hydrate_schedule_state	bool	If True, hydrate each dataset by ID and stream metadata so schedule-dependent properties (is_manual, is_scheduled) are reliable.	False
hydrate_concurrency	int	Max concurrent dataset hydrations when hydrate_schedule_state=True	10
debug_api	bool	Enable API debugging	False
debug_num_stacks_to_drop	int	Stack frames to drop for debugging	2
session	AsyncClient | None	HTTP client session (optional, will create one with 30s timeout if not provided)	None
return_raw	bool	Return raw API response	False
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
list[DomoDataset]	List of DomoDataset objects matching the criteria

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

async def get(
    self,
    search_text: str | None = None,
    maximum: int | None = None,
    data_provider_type: str | None = None,
    hydrate_schedule_state: bool = False,
    hydrate_concurrency: int = 10,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoDataset]:
    """Retrieve datasets from Domo, optionally filtering by search text.

    Args:
        search_text: Optional text to search for in dataset names
        maximum: Maximum number of datasets to return
        data_provider_type: Optional dataset provider type to filter on
        hydrate_schedule_state: If True, hydrate each dataset by ID and stream metadata
            so schedule-dependent properties (`is_manual`, `is_scheduled`) are reliable.
        hydrate_concurrency: Max concurrent dataset hydrations when
            hydrate_schedule_state=True
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop for debugging
        session: HTTP client session (optional, will create one with 30s timeout if not provided)
        return_raw: Return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        List of DomoDataset objects matching the criteria
    """
    # Create session with longer timeout if not provided
    # Datacenter search API can be slow for large result sets
    close_session = False
    if session is None:
        timeout = httpx.Timeout(30.0, connect=10.0)
        session = httpx.AsyncClient(timeout=timeout, verify=False)
        close_session = True

    try:
        context = RouteContext.build_context(
            context=context,
            session=session,
            debug_api=debug_api,
            debug_num_stacks_to_drop=debug_num_stacks_to_drop,
            **context_kwargs,
        )

        res = await dataset_routes.search_datasets(
            auth=self.auth,
            search_text=search_text,
            maximum=maximum,
            data_provider_type=data_provider_type,
            context=context,
        )

        if return_raw:
            return res

        self.datasets = self._to_domo_dataset(
            dataset_ls=res.response or [], auth=self.auth
        )

        if hydrate_schedule_state and self.datasets:
            semaphore = asyncio.Semaphore(max(1, hydrate_concurrency))

            async def _hydrate(dataset: DomoDataset):
                async with semaphore:
                    await dataset.hydrate_schedule_state(context=context)

            await asyncio.gather(*(_hydrate(ds) for ds in self.datasets))

        return self.datasets
    finally:
        if close_session:
            await session.aclose()

get_by_ids async

get_by_ids(
    dataset_ids: list[str],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoDataset]

Fetch many datasets at once (POST /data/v3/datasources/bulk).

Each result is dispatched to the right class (DomoDataset / DomoDatasetView / DomoSemanticModel) by the factory.

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

async def get_by_ids(
    self,
    dataset_ids: list[str],
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoDataset]:
    """Fetch many datasets at once (POST /data/v3/datasources/bulk).

    Each result is dispatched to the right class (DomoDataset /
    DomoDatasetView / DomoSemanticModel) by the factory.
    """
    context = RouteContext.build_context(context=context, **context_kwargs)
    res = await dataset_routes.get_datasets_by_ids(
        auth=self.auth, dataset_ids=dataset_ids, context=context
    )
    if return_raw:
        return res
    self.datasets = self._to_domo_dataset(
        dataset_ls=res.response or [], auth=self.auth
    )
    return self.datasets

search async

search(
    search_text: str,
    maximum: int | None = None,
    data_provider_type: str | None = None,
    hydrate_schedule_state: bool = False,
    hydrate_concurrency: int = 10,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoDataset]

Search for datasets by name.

Parameters:

Name	Type	Description	Default
search_text	str	Text to search for in dataset names (wildcards supported)	required
maximum	int | None	Maximum number of datasets to return	None
data_provider_type	str | None	Optional dataset provider type to filter on	None
hydrate_schedule_state	bool	If True, hydrate each dataset by ID and stream metadata so schedule-dependent properties (is_manual, is_scheduled) are reliable.	False
hydrate_concurrency	int	Max concurrent dataset hydrations when hydrate_schedule_state=True	10
debug_api	bool	Enable API debugging	False
debug_num_stacks_to_drop	int	Stack frames to drop for debugging	2
session	AsyncClient | None	HTTP client session (optional, will create one with 30s timeout if not provided)	None
return_raw	bool	Return raw API response	False
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
list[DomoDataset]	List of DomoDataset objects matching the search criteria

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

async def search(
    self,
    search_text: str,
    maximum: int | None = None,
    data_provider_type: str | None = None,
    hydrate_schedule_state: bool = False,
    hydrate_concurrency: int = 10,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoDataset]:
    """Search for datasets by name.

    Args:
        search_text: Text to search for in dataset names (wildcards supported)
        maximum: Maximum number of datasets to return
        data_provider_type: Optional dataset provider type to filter on
        hydrate_schedule_state: If True, hydrate each dataset by ID and stream metadata
            so schedule-dependent properties (`is_manual`, `is_scheduled`) are reliable.
        hydrate_concurrency: Max concurrent dataset hydrations when
            hydrate_schedule_state=True
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop for debugging
        session: HTTP client session (optional, will create one with 30s timeout if not provided)
        return_raw: Return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        List of DomoDataset objects matching the search criteria
    """
    return await self.get(
        search_text=search_text,
        maximum=maximum,
        data_provider_type=data_provider_type,
        hydrate_schedule_state=hydrate_schedule_state,
        hydrate_concurrency=hydrate_concurrency,
        debug_api=debug_api,
        debug_num_stacks_to_drop=debug_num_stacks_to_drop,
        session=session,
        return_raw=return_raw,
        context=context,
        **context_kwargs,
    )

search_by_tags async

search_by_tags(
    tags: list[str] | str,
    maximum: int | None = None,
    search_text: str | None = None,
    hydrate_schedule_state: bool = False,
    hydrate_concurrency: int = 10,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoDataset]

Search for datasets by tag(s).

Uses the datacenter search API to find datasets matching the given tag filter(s). A single tag string or a list of tag strings may be passed.

Parameters:

Name	Type	Description	Default
tags	list[str] | str	Tag filter(s) — a single tag string or a list of tag strings	required
maximum	int | None	Maximum number of datasets to return	None
search_text	str | None	Optional text to search for in dataset names (wildcards supported)	None
hydrate_schedule_state	bool	If True, hydrate each dataset by ID and stream metadata so schedule-dependent properties (is_manual, is_scheduled) are reliable.	False
hydrate_concurrency	int	Max concurrent dataset hydrations when hydrate_schedule_state=True	10
debug_api	bool	Enable API debugging	False
debug_num_stacks_to_drop	int	Stack frames to drop for debugging	2
session	AsyncClient | None	HTTP client session (optional, will create one with 30s timeout if not provided)	None
return_raw	bool	Return raw API response	False
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
list[DomoDataset]	List of DomoDataset objects matching the tag criteria

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

async def search_by_tags(
    self,
    tags: list[str] | str,
    maximum: int | None = None,
    search_text: str | None = None,
    hydrate_schedule_state: bool = False,
    hydrate_concurrency: int = 10,
    debug_api: bool = False,
    debug_num_stacks_to_drop: int = 2,
    session: httpx.AsyncClient | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoDataset]:
    """Search for datasets by tag(s).

    Uses the datacenter search API to find datasets matching the given tag
    filter(s). A single tag string or a list of tag strings may be passed.

    Args:
        tags: Tag filter(s) — a single tag string or a list of tag strings
        maximum: Maximum number of datasets to return
        search_text: Optional text to search for in dataset names (wildcards supported)
        hydrate_schedule_state: If True, hydrate each dataset by ID and stream metadata
            so schedule-dependent properties (`is_manual`, `is_scheduled`) are reliable.
        hydrate_concurrency: Max concurrent dataset hydrations when
            hydrate_schedule_state=True
        debug_api: Enable API debugging
        debug_num_stacks_to_drop: Stack frames to drop for debugging
        session: HTTP client session (optional, will create one with 30s timeout if not provided)
        return_raw: Return raw API response
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        List of DomoDataset objects matching the tag criteria
    """
    # Create session with longer timeout if not provided
    # Datacenter search API can be slow for large result sets
    close_session = False
    if session is None:
        timeout = httpx.Timeout(30.0, connect=10.0)
        session = httpx.AsyncClient(timeout=timeout, verify=False)
        close_session = True

    try:
        context = RouteContext.build_context(
            context=context,
            session=session,
            debug_api=debug_api,
            debug_num_stacks_to_drop=debug_num_stacks_to_drop,
            **context_kwargs,
        )

        res = await dataset_routes.search_datasets(
            auth=self.auth,
            search_text=search_text,
            maximum=maximum,
            tags=tags,
            context=context,
        )

        if return_raw:
            return res

        self.datasets = self._to_domo_dataset(
            dataset_ls=res.response or [], auth=self.auth
        )

        if hydrate_schedule_state and self.datasets:
            semaphore = asyncio.Semaphore(max(1, hydrate_concurrency))

            async def _hydrate(dataset: DomoDataset):
                async with semaphore:
                    await dataset.hydrate_schedule_state(context=context)

            await asyncio.gather(*(_hydrate(ds) for ds in self.datasets))

        return self.datasets
    finally:
        if close_session:
            await session.aclose()