access_controller

access_controller

Card access control using the relationship system.

DomoCardAccessController

DomoCardAccessController(auth: DomoAuth, parent)

Bases: DomoAccessRelationshipController

Access controller for Domo cards.

Source code in src/crew_dcs/classes/DomoCard/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,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> bool

Add owners to the card.

Source code in src/crew_dcs/classes/DomoCard/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,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Add owners to the card."""
    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")

    existing = await self._get_owner_entries(context=context)
    additions = [{"id": str(uid), "type": "USER"} for uid in user_ids] + [
        {"id": str(gid), "type": "GROUP"} for gid in group_ids
    ]

    combined = {f"{o['type']}:{o['id']}": o for o in existing}
    combined.update({f"{o['type']}:{o['id']}": o for o in additions})

    res = await card_routes.replace_card_owners(
        auth=self.auth,
        card_id=self.parent_id,
        owners=list(combined.values()),
        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 card.

Source code in src/crew_dcs/classes/DomoCard/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 card."""
    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 card as relationships.

Source code in src/crew_dcs/classes/DomoCard/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 card as relationships."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await card_routes.get_card_access_list(
        auth=self.auth,
        card_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.CARD,
                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.CARD,
                relationship_type=self._map_access_level(
                    group.get("accessLevel")
                    or group.get("permission")
                    or group.get("access")
                ),
                metadata=group,
            )
        )

    owner_relationships = await self.get_owners(context=context)

    for owner_rel in owner_relationships:
        existing = next(
            (
                r
                for r in relationships
                if r.from_entity_id == owner_rel.from_entity_id
                and r.from_entity_type == owner_rel.from_entity_type
            ),
            None,
        )
        if existing:
            existing.relationship_type = RelationshipType.HAS_ACCESS_OWNER
            existing.metadata["is_owner"] = True
        else:
            relationships.append(owner_rel)

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

get_direct_relationships async

get_direct_relationships() -> list[Relationship]

Get direct relationships for the card.

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

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

get_owner_entities async

get_owner_entities(
    is_suppress_errors: bool = True,
    *,
    context: RouteContext | None = None,
    **context_kwargs
) -> list[Any]

Resolve owner entities (users/groups) for this card.

Parameters:

Name	Type	Description	Default
is_suppress_errors	bool	When True, suppress DomoError resolution failures.	True
context	RouteContext | None	Optional RouteContext for API call configuration	None
**context_kwargs		Additional context parameters	{}

Returns:

Type	Description
list[Any]	list of DomoUser/DomoGroup entities

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

async def get_owner_entities(
    self,
    is_suppress_errors: bool = True,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Any]:
    """Resolve owner entities (users/groups) for this card.

    Args:
        is_suppress_errors: When True, suppress DomoError resolution failures.
        context: Optional RouteContext for API call configuration
        **context_kwargs: Additional context parameters

    Returns:
        list of DomoUser/DomoGroup entities
    """
    from ..DomoGroup.core import DomoGroup
    from ..DomoUser import DomoUser

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

    relationships = await self.get_owners(context=context)

    tasks = []
    for rel in relationships:
        try:
            if rel.from_entity_type == EntityType.USER:
                tasks.append(
                    DomoUser.get_by_id(
                        auth=self.auth,
                        user_id=rel.from_entity_id,
                        context=context,
                    )
                )
            elif rel.from_entity_type == EntityType.GROUP:
                tasks.append(
                    DomoGroup.get_by_id(
                        auth=self.auth,
                        group_id=rel.from_entity_id,
                        context=context,
                    )
                )
        except DomoError:
            if not is_suppress_errors:
                raise

    if not tasks:
        return []

    return await dmce.gather_with_concurrency(n=60, *tasks)  # noqa: B026

get_owners async

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

Get owner relationships for the card.

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

async def get_owners(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get owner relationships for the card."""
    context = RouteContext.build_context(context=context, **context_kwargs)

    res = await card_routes.get_card_metadata(
        auth=self.auth,
        card_id=self.parent_id,
        optional_parts="owners",
        context=context,
    )

    owners = (
        res.response.get("owners", []) if isinstance(res.response, dict) else []
    )

    relationships: list[Relationship] = []
    for owner in owners:
        owner_id = owner.get("id")
        owner_type = owner.get("type", "USER")
        if not owner_id:
            continue
        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.CARD,
                relationship_type=RelationshipType.HAS_ACCESS_OWNER,
                metadata={"is_owner": True, **owner},
            )
        )

    return relationships

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

Remove owners from the card.

Source code in src/crew_dcs/classes/DomoCard/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,
    return_raw: bool = False,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> bool:
    """Remove owners from the card."""
    context = RouteContext.build_context(context=context, **context_kwargs)

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

    existing = await self._get_owner_entries(context=context)
    remaining = [
        owner
        for owner in existing
        if not (
            (owner.get("type") == "USER" and owner.get("id") in user_ids)
            or (owner.get("type") == "GROUP" and owner.get("id") in group_ids)
        )
    ]

    if not remaining:
        raise ValueError("Cannot remove all owners; owners would be empty")

    res = await card_routes.replace_card_owners(
        auth=self.auth,
        card_id=self.parent_id,
        owners=remaining,
        context=context,
    )

    return res if return_raw else res.is_success

remove_relationship async

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

Remove an access relationship for the card.

Source code in src/crew_dcs/classes/DomoCard/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 card."""
    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 card 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 card with users or groups.

Source code in src/crew_dcs/classes/DomoCard/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 card 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 card")

    res = await share_resource(
        auth=self.auth,
        resource_ids=self.parent_id,
        resource_type=ShareResource_Enum.CARD,
        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