Skip to content

sharing

sharing

Dataset sharing operations.

generate_share_dataset_payload 

generate_share_dataset_payload(
    entity_type: str,
    entity_id: str,
    access_level: ShareDataset_AccessLevelEnum = CAN_SHARE,
    is_send_email: bool = False,
) -> dict

Generate payload for sharing a dataset.

Parameters:

Name Type Description Default
entity_type str

Type of entity (USER or GROUP)

 required
entity_id str

ID of the user or group

 required
access_level ShareDataset_AccessLevelEnum

Access level (ShareDataset_AccessLevelEnum enum). Defaults to CAN_SHARE.

 CAN_SHARE
is_send_email bool

Whether to send email notification

 False

Returns:

Type Description
dict

Dictionary payload for dataset sharing API
Source code in src/crew_dcs/routes/dataset/sharing.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def generate_share_dataset_payload(
    entity_type: str,  # USER or GROUP
    entity_id: str,
    access_level: ShareDataset_AccessLevelEnum = ShareDataset_AccessLevelEnum.CAN_SHARE,
    is_send_email: bool = False,
) -> dict:
    """Generate payload for sharing a dataset.

    Args:
        entity_type: Type of entity (USER or GROUP)
        entity_id: ID of the user or group
        access_level: Access level (ShareDataset_AccessLevelEnum enum). Defaults to CAN_SHARE.
        is_send_email: Whether to send email notification

    Returns:
        Dictionary payload for dataset sharing API
    """
    return {
        "permissions": [
            {"type": entity_type, "id": entity_id, "accessLevel": access_level.value}
        ],
        "sendEmail": is_send_email,
    }

get_permissions async 

get_permissions(
    auth: DomoAuth,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> ResponseGetData

retrieve the schema for a dataset

Source code in src/crew_dcs/routes/dataset/sharing.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@gd.route_function
@log_call(
    level_name="route",
    config=LogDecoratorConfig(
        entity_extractor=DomoEntityExtractor(),
        result_processor=DomoEntityResultProcessor(),
    ),
)
async def get_permissions(
    auth: DomoAuth,
    dataset_id: str,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> rgd.ResponseGetData:
    """retrieve the schema for a dataset"""

    url = f"https://{auth.domo_instance}.domo.com/api/data/v3/datasources/{dataset_id}/permissions"

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

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

    return res