Skip to content

access_controller

access_controller

Dataset access control using the relationship system.

DomoDatasetAccessController 

DomoDatasetAccessController(auth: DomoAuth, parent)

Bases: DomoAccessRelationshipController

Access controller for Domo datasets.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
65
66
67
68
def __init__(self, auth: DomoAuth, parent):
    super().__init__(auth=auth, parent_object=parent)
    self.parent = parent
    self.relationships: list[Relationship] = []

add_owners async 

add_owners(
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    is_send_email: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Add owners to the dataset.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
async def add_owners(
    self,
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    is_send_email: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Add owners to the dataset."""
    return await self.share(
        domo_users=domo_users,
        domo_groups=domo_groups,
        user_ids=user_ids,
        group_ids=group_ids,
        access_level=ShareDataset_AccessLevelEnum.CO_OWNER,
        is_send_email=is_send_email,
        return_raw=return_raw,
        context=context,
        **context_kwargs,
    )

create_relationship async 

create_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Create an access relationship for the dataset.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Create an access relationship for the dataset."""
    access_level = self._map_relationship_to_dataset_access(relationship_type)
    return await self.share(
        user_ids=(
            [target_entity_id] if target_entity_type == EntityType.USER else None
        ),
        group_ids=(
            [target_entity_id] if target_entity_type == EntityType.GROUP else None
        ),
        access_level=access_level,
        **kwargs,
    )

get_accesslist async 

get_accesslist(
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Relationship]

Get access list for the dataset as relationships.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
async def get_accesslist(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get access list for the dataset as relationships."""
    context = RouteContext.build_context(context=context, **context_kwargs)

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

    if return_raw:
        return res

    relationships: list[Relationship] = []
    permissions: list[dict[str, Any]] = []

    if isinstance(res.response, dict):
        if isinstance(res.response.get("permissions"), list):
            permissions = res.response.get("permissions", [])
        elif isinstance(res.response.get("accessList"), list):
            permissions = res.response.get("accessList", [])
        elif isinstance(res.response.get("accessList"), dict):
            permissions = res.response.get("accessList", {}).get("permissions", [])
    elif isinstance(res.response, list):
        permissions = res.response

    for perm in permissions:
        entity_id = perm.get("id") or perm.get("entityId")
        entity_type = perm.get("type") or perm.get("entityType")
        access_level = perm.get("accessLevel") or perm.get("access")
        if not entity_id or not entity_type:
            continue
        relationships.append(
            Relationship(
                from_entity_id=str(entity_id),
                from_entity_type=(
                    EntityType.GROUP
                    if str(entity_type).upper() == "GROUP"
                    else EntityType.USER
                ),
                to_entity_id=str(self.parent_id),
                to_entity_type=EntityType.DATASET,
                relationship_type=self._map_access_level(access_level),
                metadata=perm,
            )
        )

    owner = getattr(self.parent, "owner", None)
    if isinstance(owner, dict) and owner.get("id"):
        relationships.append(
            Relationship(
                from_entity_id=str(owner.get("id")),
                from_entity_type=EntityType.USER,
                to_entity_id=str(self.parent_id),
                to_entity_type=EntityType.DATASET,
                relationship_type=RelationshipType.HAS_ACCESS_OWNER,
                metadata={"is_owner": True, **owner},
            )
        )

    self.relationships = relationships
    self.invalidate_cache()
    return relationships

get_direct_relationships async 

get_direct_relationships() -> list[Relationship]

Get direct relationships for the dataset.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
74
75
76
async def get_direct_relationships(self) -> list[Relationship]:
    """Get direct relationships for the dataset."""
    return await self.get_accesslist()

get_owners async 

get_owners(
    *, context: RouteContext | None = None, **context_kwargs
) -> list[Relationship]

Get owner relationships for the dataset.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
async def get_owners(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get owner relationships for the dataset."""
    if not self.relationships:
        await self.get_accesslist(context=context, **context_kwargs)

    return [
        rel
        for rel in self.relationships
        if rel.relationship_type == RelationshipType.HAS_ACCESS_OWNER
    ]

remove_owners async 

remove_owners(
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    **_kwargs
) -> bool

Remove owners from the dataset.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
221
222
223
224
225
226
227
228
229
230
231
232
async def remove_owners(
    self,
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    **_kwargs,
) -> bool:
    """Remove owners from the dataset."""
    raise NotImplementedError(
        "Removing dataset owners is not supported by current routes."
    )

remove_relationship async 

remove_relationship(
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs
) -> bool

Remove an access relationship for the dataset.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
 98
 99
100
101
102
103
104
105
106
107
108
async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove an access relationship for the dataset."""
    raise NotImplementedError(
        "Removing dataset share relationships is not supported by current routes."
    )

share async 

share(
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    access_level: ShareDataset_AccessLevelEnum = CAN_SHARE,
    is_send_email: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Share the dataset with users or groups.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
async def share(
    self,
    domo_users: list[Any] | None = None,
    domo_groups: list[Any] | None = None,
    user_ids: list[str | int] | None = None,
    group_ids: list[str | int] | None = None,
    access_level: ShareDataset_AccessLevelEnum = ShareDataset_AccessLevelEnum.CAN_SHARE,
    is_send_email: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Share the dataset with users or groups."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    user_ids = self._collect_ids(domo_users, user_ids)
    group_ids = self._collect_ids(domo_groups, group_ids)

    permissions = [
        {"type": "USER", "id": str(uid), "accessLevel": access_level.value}
        for uid in user_ids
    ] + [
        {"type": "GROUP", "id": str(gid), "accessLevel": access_level.value}
        for gid in group_ids
    ]

    if not permissions:
        raise ValueError("Must provide user or group identifiers to share dataset")

    body = {"permissions": permissions, "sendEmail": is_send_email}

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

    return res if return_raw else res.is_success

ShareDataset_UnsupportedAccessLevel 

ShareDataset_UnsupportedAccessLevel(
    relationship_type: RelationshipType,
    supported: list[str] | None = None,
    cls_instance: Any = None,
)

Bases: ClassError

Raised when a caller asks for a dataset access level Domo cannot grant.

Domo's dataset share API has exactly three levels — CO_OWNER, CAN_EDIT and CAN_SHARE (routes/dataset/sharing.py:ShareDataset_AccessLevelEnum). There is no read-only dataset share.

Refusing here is deliberate. This mapping used to resolve the read-only relationship types (and its fallback) UP to CAN_SHARE, so asking for "viewer" silently granted the recipient the right to re-share the dataset to anyone. RelationshipType.default is HAS_ACCESS_VIEWER, so the default path escalated too. Verified live against domo-community on 2026-08-31: a HAS_ACCESS_VIEWER grant landed as "accessLevel": "CAN_SHARE".

A privilege grant must never be quietly widened to fit the API. Name the levels the entity actually supports and let the caller choose. See .agents/guides/access-and-sharing.md.

Source code in src/crew_dcs/classes/DomoDataset/access_controller.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(
    self,
    relationship_type: RelationshipType,
    supported: list[str] | None = None,
    cls_instance: Any = None,
):
    self.relationship_type = relationship_type
    self.supported = supported or []
    message = (
        f"cannot grant dataset access for relationship_type "
        f"{relationship_type.name}: Domo datasets have no equivalent level. "
        f"Supported levels are {', '.join(self.supported)} — pass one "
        f"explicitly (Domo has no read-only dataset share). Widening the "
        f"request to CAN_SHARE would grant re-share rights."
    )
    super().__init__(cls_instance=cls_instance, message=message)