schema

schema

a class based approach for interacting with Domo Datasets

DomoDataset_Schema dataclass

DomoDataset_Schema(
    parent: DomoEntity,
    columns: list[DomoDataset_Schema_Column] = list(),
)

Bases: DomoSubEntity

class for interacting with dataset schemas

add_col async

add_col(
    col: DomoDataset_Schema_Column,
    debug_prn: bool = False,
    is_update_schema: bool = True,
    debug_api: bool = False,
    session: AsyncClient = None,
)

Add a column to the schema.

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

async def add_col(
    self,
    col: DomoDataset_Schema_Column,
    debug_prn: bool = False,
    is_update_schema: bool = True,
    debug_api: bool = False,
    session: httpx.AsyncClient = None,
):
    """Add a column to the schema."""
    if col in self.columns and debug_prn:
        print(
            f"column - {col.name} already in dataset {self.parent.name if self.parent else ''}"
        )

    if col not in self.columns:
        self.columns.append(col)

    if not is_update_schema:
        return self.columns

    context = RouteContext.build_context(
        session=session,
        debug_api=debug_api,
    )
    return await self.update_schema(context=context)

get async

get(
    debug_api: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[DomoDataset_Schema_Column]

method that retrieves schema for a dataset

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

async def get(
    self,
    debug_api: bool = False,
    return_raw: bool = False,  # return the raw response
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[DomoDataset_Schema_Column]:
    """method that retrieves schema for a dataset"""

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

    res = await dataset_routes.get_schema(
        auth=self.auth, dataset_id=self.parent_id, context=context
    )

    self.raw = res.response

    if return_raw:
        return res

    self.columns = [
        DomoDataset_Schema_Column.from_dict(obj=obj)
        for obj in res.response.get("tables")[0].get("columns")
    ]

    return self.columns

remove_col async

remove_col(
    col_to_remove: DomoDataset_Schema_Column,
    debug_api: bool = False,
    is_update_schema: bool = True,
    session: AsyncClient = None,
)

Remove a column from the schema.

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

async def remove_col(
    self,
    col_to_remove: "DomoDataset_Schema_Column",
    debug_api: bool = False,
    is_update_schema: bool = True,
    session: httpx.AsyncClient = None,
):
    """Remove a column from the schema."""
    [
        self.columns.pop(index)
        for index, col in enumerate(self.columns)
        if col == col_to_remove
    ]

    if not is_update_schema:
        return self.columns

    context = RouteContext.build_context(
        session=session,
        debug_api=debug_api,
    )
    return await self.update_schema(context=context)