access_controller

access_controller

Page access control using the relationship system.

DomoPageAccessController

DomoPageAccessController(auth: DomoAuth, parent)

Bases: DomoAccessRelationshipController

Access controller for Domo pages.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

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,
    message: str | None = None,
    send_email: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Add owners to the page.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

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,
    message: str | None = None,
    send_email: bool = False,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Add owners to the page."""
    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)

    if not user_ids and not group_ids:
        raise ValueError("Must provide user or group identifiers to add owners")

    res = await page_routes.add_page_owner(
        auth=self.auth,
        page_id_ls=[self.parent_id],
        group_id_ls=group_ids,
        user_id_ls=user_ids,
        note=message or "",
        send_email=send_email,
        return_raw=return_raw,
        context=context,
    )

    return res if return_raw else res.is_success

create_relationship async

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

Create an access relationship for the page.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Create an access relationship for the page."""
    if relationship_type == RelationshipType.HAS_ACCESS_OWNER:
        return await self.add_owners(
            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
            ),
            **kwargs,
        )

    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
        ),
        **kwargs,
    )

get_accesslist async

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

Get access list for the page as relationships.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

async def get_accesslist(
    self,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get access list for the page as relationships."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await page_routes.get_page_access_list(
        auth=self.auth,
        page_id=self.parent_id,
        is_expand_users=True,
        return_raw=return_raw,
        context=context,
    )

    if return_raw:
        return res

    relationships: list[Relationship] = []
    users = res.response.get("users", []) if isinstance(res.response, dict) else []
    groups = (
        res.response.get("groups", []) if isinstance(res.response, dict) else []
    )

    for user in users:
        user_id = user.get("id") or user.get("userId")
        if not user_id:
            continue
        relationships.append(
            Relationship(
                from_entity_id=str(user_id),
                from_entity_type=EntityType.USER,
                to_entity_id=str(self.parent_id),
                to_entity_type=EntityType.PAGE,
                relationship_type=self._map_access_level(
                    user.get("accessLevel")
                    or user.get("permission")
                    or user.get("access")
                ),
                metadata={
                    "is_explicit_share": user.get("isExplicitShare", False),
                    **user,
                },
            )
        )

    for group in groups:
        group_id = group.get("id") or group.get("groupId")
        if not group_id:
            continue
        relationships.append(
            Relationship(
                from_entity_id=str(group_id),
                from_entity_type=EntityType.GROUP,
                to_entity_id=str(self.parent_id),
                to_entity_type=EntityType.PAGE,
                relationship_type=self._map_access_level(
                    group.get("accessLevel")
                    or group.get("permission")
                    or group.get("access")
                ),
                metadata=group,
            )
        )

    owner_res = await page_routes.get_page_access_test(
        auth=self.auth,
        page_id=self.parent_id,
        context=context,
    )
    owners = (
        owner_res.response.get("owners", [])
        if isinstance(owner_res.response, dict)
        else []
    )

    for owner in owners:
        owner_id = owner.get("id")
        owner_type = owner.get("type", "USER")
        if not owner_id:
            continue
        existing = next(
            (
                r
                for r in relationships
                if r.from_entity_id == str(owner_id)
                and r.from_entity_type
                == (EntityType.GROUP if owner_type == "GROUP" else EntityType.USER)
            ),
            None,
        )
        if existing:
            existing.relationship_type = RelationshipType.HAS_ACCESS_OWNER
            existing.metadata["is_owner"] = True
        else:
            relationships.append(
                Relationship(
                    from_entity_id=str(owner_id),
                    from_entity_type=(
                        EntityType.GROUP
                        if owner_type == "GROUP"
                        else EntityType.USER
                    ),
                    to_entity_id=str(self.parent_id),
                    to_entity_type=EntityType.PAGE,
                    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 page.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

async def get_direct_relationships(self) -> list[Relationship]:
    """Get direct relationships for the page."""
    return await self.get_accesslist()

get_owners async

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

Get owner relationships for the page.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

async def get_owners(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get owner relationships for the page."""
    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 page.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

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 page."""
    raise NotImplementedError(
        "Removing page owners is not supported by current page 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 page.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove an access relationship for the page."""
    if relationship_type == RelationshipType.HAS_ACCESS_OWNER:
        return await self.remove_owners(
            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
            ),
            **kwargs,
        )

    raise NotImplementedError(
        "Removing page 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,
    message: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Share the page with users or groups.

Source code in src/crew_dcs/classes/DomoPage/access_controller.py

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,
    message: str | None = None,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Share the page 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)

    if not user_ids and not group_ids:
        raise ValueError("Must provide user or group identifiers to share page")

    res = await share_resource(
        auth=self.auth,
        resource_ids=self.parent_id,
        resource_type=ShareResource_Enum.PAGE,
        group_ids=group_ids,
        user_ids=user_ids,
        message=message,
        return_raw=return_raw,
        context=context,
    )

    return res if return_raw else res.is_success