metadata

metadata

check_for_colliding_datasources async

check_for_colliding_datasources(
    auth: DomoAuth,
    dataset_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Check for Cloud Amplifier integrations that collide with an existing Domo Dataset.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
dataset_id	str	Dataset ID to check for collisions	required
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing collision information

Raises:

Type	Description
CloudAmplifier_GET_Error	If collision check fails or no federated metadata exists

Example

collisions = await check_for_colliding_datasources(auth, "dataset-123") print(collisions.response)

Source code in src/crew_dcs/routes/cloud_amplifier/metadata.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def check_for_colliding_datasources(
    auth: DomoAuth,
    dataset_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Check for Cloud Amplifier integrations that collide with an existing Domo Dataset.

    Args:
        auth: Authentication object containing instance and credentials
        dataset_id: Dataset ID to check for collisions
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing collision information

    Raises:
        CloudAmplifier_GET_Error: If collision check fails or no federated metadata exists

    Example:
        >>> collisions = await check_for_colliding_datasources(auth, "dataset-123")
        >>> print(collisions.response)
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/migration/integrations/datasource/{dataset_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if return_raw:
        return res

    # A 400 may be returned if no federated metadata exists for the dataset
    if res.status == 400:
        raise CloudAmplifier_GET_Error(
            entity_id=dataset_id,
            res=res,
            message=f"No federated metadata exists for the datasource {dataset_id}",
        )

    if not res.is_success:
        raise CloudAmplifier_GET_Error(entity_id=dataset_id, res=res)

    return res

get_databases async

get_databases(
    auth: DomoAuth,
    integration_id: str,
    page: int = 0,
    rows: int = 5000,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve a list of all databases for a Cloud Amplifier integration.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
integration_id	str	Integration ID to list databases for	required
page	int	Page number for pagination (default: 0)	0
rows	int	Number of rows per page (default: 5000)	5000
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing list of databases

Raises:

Type	Description
CloudAmplifier_GET_Error	If database retrieval fails

Example

databases = await get_databases(auth, "integration-123") for db in databases.response: ... print(db['databaseName'])

Source code in src/crew_dcs/routes/cloud_amplifier/metadata.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_databases(
    auth: DomoAuth,
    integration_id: str,
    page: int = 0,
    rows: int = 5000,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieve a list of all databases for a Cloud Amplifier integration.

    Args:
        auth: Authentication object containing instance and credentials
        integration_id: Integration ID to list databases for
        page: Page number for pagination (default: 0)
        rows: Number of rows per page (default: 5000)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing list of databases

    Raises:
        CloudAmplifier_GET_Error: If database retrieval fails

    Example:
        >>> databases = await get_databases(auth, "integration-123")
        >>> for db in databases.response:
        ...     print(db['databaseName'])
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/byos/accounts/{integration_id}/databases"

    params = {"page": page, "rows": rows}

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        params=params,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise CloudAmplifier_GET_Error(entity_id=integration_id, res=res)

    return res

get_federated_source_metadata async

get_federated_source_metadata(
    auth: DomoAuth,
    dataset_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve federated source metadata for a dataset.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
dataset_id	str	Dataset ID to retrieve metadata for	required
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing federated source metadata

Raises:

Type	Description
SearchCloudAmplifierNotFoundError	If no federated datasource exists with the ID
CloudAmplifier_GET_Error	If metadata retrieval fails

Example

metadata = await get_federated_source_metadata(auth, "dataset-123") print(metadata.response)

Source code in src/crew_dcs/routes/cloud_amplifier/metadata.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_federated_source_metadata(
    auth: DomoAuth,
    dataset_id: str,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieve federated source metadata for a dataset.

    Args:
        auth: Authentication object containing instance and credentials
        dataset_id: Dataset ID to retrieve metadata for
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing federated source metadata

    Raises:
        SearchCloudAmplifierNotFoundError: If no federated datasource exists with the ID
        CloudAmplifier_GET_Error: If metadata retrieval fails

    Example:
        >>> metadata = await get_federated_source_metadata(auth, "dataset-123")
        >>> print(metadata.response)
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/federated/v1/config/datasources/{dataset_id}"

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        context=context,
    )

    if return_raw:
        return res

    # A 404 may be returned if no federated metadata exists for the dataset
    if res.status == 404:
        raise SearchCloudAmplifierNotFoundError(
            search_criteria=f"Federated datasource ID: {dataset_id}",
            res=res,
        )

    if not res.is_success:
        raise CloudAmplifier_GET_Error(entity_id=dataset_id, res=res)

    return res

get_schemas async

get_schemas(
    auth: DomoAuth,
    integration_id: str,
    database: str,
    page: int = 0,
    rows: int = 5000,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve a list of all schemas for a Cloud Amplifier integration database.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
integration_id	str	Integration ID	required
database	str	Database name to list schemas for	required
page	int	Page number for pagination (default: 0)	0
rows	int	Number of rows per page (default: 5000)	5000
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing list of schemas

Raises:

Type	Description
CloudAmplifier_GET_Error	If schema retrieval fails

Example

schemas = await get_schemas(auth, "integration-123", "MY_DATABASE") for schema in schemas.response: ... print(schema['schemaName'])

Source code in src/crew_dcs/routes/cloud_amplifier/metadata.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_schemas(
    auth: DomoAuth,
    integration_id: str,
    database: str,
    page: int = 0,
    rows: int = 5000,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieve a list of all schemas for a Cloud Amplifier integration database.

    Args:
        auth: Authentication object containing instance and credentials
        integration_id: Integration ID
        database: Database name to list schemas for
        page: Page number for pagination (default: 0)
        rows: Number of rows per page (default: 5000)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing list of schemas

    Raises:
        CloudAmplifier_GET_Error: If schema retrieval fails

    Example:
        >>> schemas = await get_schemas(auth, "integration-123", "MY_DATABASE")
        >>> for schema in schemas.response:
        ...     print(schema['schemaName'])
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/byos/accounts/{integration_id}/databases/{database}/schemas"

    params = {"page": page, "rows": rows}

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        params=params,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise CloudAmplifier_GET_Error(entity_id=integration_id, res=res)

    return res

get_tables async

get_tables(
    auth: DomoAuth,
    integration_id: str,
    database: str,
    schema: str,
    page: int = 0,
    rows: int = 5000,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

Retrieve a list of all tables for a Cloud Amplifier integration schema.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object containing instance and credentials	required
integration_id	str	Integration ID	required
database	str	Database name	required
schema	str	Schema name to list tables for	required
page	int	Page number for pagination (default: 0)	0
rows	int	Number of rows per page (default: 5000)	5000
return_raw	bool	Return raw API response without processing	False
context	RouteContext | None	RouteContext for request configuration	None
**context_kwargs		Additional context parameters (session, debug_api, etc.)	{}

Returns:

Type	Description
ResponseGetData	ResponseGetData object containing list of tables

Raises:

Type	Description
CloudAmplifier_GET_Error	If table retrieval fails

Example

tables = await get_tables(auth, "integration-123", "MY_DB", "MY_SCHEMA") for table in tables.response: ... print(table['tableName'])

Source code in src/crew_dcs/routes/cloud_amplifier/metadata.py

@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_tables(
    auth: DomoAuth,
    integration_id: str,
    database: str,
    schema: str,
    page: int = 0,
    rows: int = 5000,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """
    Retrieve a list of all tables for a Cloud Amplifier integration schema.

    Args:
        auth: Authentication object containing instance and credentials
        integration_id: Integration ID
        database: Database name
        schema: Schema name to list tables for
        page: Page number for pagination (default: 0)
        rows: Number of rows per page (default: 5000)
        return_raw: Return raw API response without processing
        context: RouteContext for request configuration
        **context_kwargs: Additional context parameters (session, debug_api, etc.)

    Returns:
        ResponseGetData object containing list of tables

    Raises:
        CloudAmplifier_GET_Error: If table retrieval fails

    Example:
        >>> tables = await get_tables(auth, "integration-123", "MY_DB", "MY_SCHEMA")
        >>> for table in tables.response:
        ...     print(table['tableName'])
    """
    context = RouteContext.build_context(context=context, **context_kwargs)

    url = f"https://{auth.domo_instance}.domo.com/api/query/v1/byos/accounts/{integration_id}/databases/{database}/schemas/{schema}/objects"

    params = {"page": page, "rows": rows}

    res = await gd.get_data(
        auth=auth,
        url=url,
        method="GET",
        params=params,
        context=context,
    )

    if return_raw:
        return res

    if not res.is_success:
        raise CloudAmplifier_GET_Error(entity_id=integration_id, res=res)

    return res