Relationships

Relationships

Unified Relationship System for Domo Objects

This module provides a comprehensive relationship framework that models all types of connections between Domo entities including access control, group membership, lineage tracking, and certification.

Key Concepts: 1. Relationships: Directional connections between entities (A → B) 2. Relationship Types: The nature of the connection (access, membership, lineage, etc.) 3. Entity Types: What kinds of objects can have relationships (users, groups, datasets, etc.) 4. Relationship Controllers: Manage specific types of relationships for objects

Examples of Relationships: - User HAS_ACCESS_EDITOR → Card - User IS_MEMBER_OF → Group - Dataset IS_DERIVED_FROM → Dataset - Card IS_CERTIFIED_BY → User - Dataset IS_TAGGED_WITH → Tag

DomoAccessRelationshipController

DomoAccessRelationshipController(
    auth: DomoAuth, parent_object: DomoEntity
)

Bases: DomoRelationshipController

Relationship controller for access/sharing relationships.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def __init__(self, auth: DomoAuth, parent_object: DomoEntity):
    self.auth = auth
    self.parent_object = parent_object
    self.parent_id = parent_object.id

    # Cache for relationship data
    self._relationship_cache: dict[str, RelationshipSummary] = {}
    self._cache_valid = False

create_relationship async

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

Create an access relationship.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Create an access relationship."""

get_direct_relationships async

get_direct_relationships() -> list[Relationship]

Get direct access relationships for the object.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def get_direct_relationships(self) -> list[Relationship]:
    """Get direct access relationships for the object."""

remove_relationship async

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

Remove an access relationship.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove an access relationship."""

DomoMembershipRelationshipController

DomoMembershipRelationshipController(
    auth: DomoAuth, parent_object: DomoEntity
)

Bases: DomoRelationshipController

Relationship controller for group membership relationships.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def __init__(self, auth: DomoAuth, parent_object: DomoEntity):
    self.auth = auth
    self.parent_object = parent_object
    self.parent_id = parent_object.id

    # Cache for relationship data
    self._relationship_cache: dict[str, RelationshipSummary] = {}
    self._cache_valid = False

create_relationship async

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

Add entity to group with specified relationship.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Add entity to group with specified relationship."""

get_direct_relationships async

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

Get direct membership relationships for the group.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def get_direct_relationships(
    self,
    *,
    context: RouteContext | None = None,
    **context_kwargs,
) -> list[Relationship]:
    """Get direct membership relationships for the group."""
    from ...routes import group as group_routes

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

    # Get owners and members
    owners_res = await group_routes.get_group_owners(
        auth=self.auth, group_id=self.parent_id, context=context
    )

    members_res = await group_routes.get_group_membership(
        auth=self.auth, group_id=self.parent_id, context=context
    )

    relationships = []

    # Process owners
    for owner in owners_res.response:
        rel = Relationship(
            from_entity_id=owner.get("userId") or owner.get("id"),
            from_entity_type=(
                EntityType.USER
                if owner.get("type", "USER") == "USER"
                else EntityType.GROUP
            ),
            to_entity_id=self.parent_id,
            to_entity_type=EntityType.GROUP,
            relationship_type=RelationshipType.IS_OWNER_OF,
        )
        relationships.append(rel)

    # Process members
    for member in members_res.response:
        rel = Relationship(
            from_entity_id=member.get("userId") or member.get("id"),
            from_entity_type=(
                EntityType.USER
                if member.get("type", "USER") == "USER"
                else EntityType.GROUP
            ),
            to_entity_id=self.parent_id,
            to_entity_type=EntityType.GROUP,
            relationship_type=RelationshipType.IS_MEMBER_OF,
        )
        relationships.append(rel)

    return relationships

remove_relationship async

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

Remove entity from group.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove entity from group."""

DomoRelationshipController

DomoRelationshipController(
    auth: DomoAuth, parent_object: DomoEntity
)

Bases: ABC

Abstract base class for managing relationships for specific object types.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def __init__(self, auth: DomoAuth, parent_object: DomoEntity):
    self.auth = auth
    self.parent_object = parent_object
    self.parent_id = parent_object.id

    # Cache for relationship data
    self._relationship_cache: dict[str, RelationshipSummary] = {}
    self._cache_valid = False

create_relationship abstractmethod async

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

Create a new relationship.

Source code in src/crew_dcs/classes/subentity/Relationships.py

@abstractmethod
async def create_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Create a new relationship."""

get_all_relationship_summaries async

get_all_relationship_summaries() -> (
    dict[str, RelationshipSummary]
)

Get relationship summaries for all entities with relationships to this object.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def get_all_relationship_summaries(self) -> dict[str, RelationshipSummary]:
    """Get relationship summaries for all entities with relationships to this object."""
    if not self._cache_valid:
        await self._refresh_relationship_cache()

    return self._relationship_cache.copy()

get_direct_relationships abstractmethod async

get_direct_relationships() -> list[Relationship]

Get all direct relationships for the parent object.

Source code in src/crew_dcs/classes/subentity/Relationships.py

@abstractmethod
async def get_direct_relationships(self) -> list[Relationship]:
    """Get all direct relationships for the parent object."""

get_relationship_summary async

get_relationship_summary(
    entity_id: str,
) -> RelationshipSummary

Get comprehensive relationship summary for a specific entity.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def get_relationship_summary(self, entity_id: str) -> RelationshipSummary:
    """Get comprehensive relationship summary for a specific entity."""
    if not self._cache_valid or entity_id not in self._relationship_cache:
        await self._refresh_relationship_cache()

    return self._relationship_cache.get(
        entity_id,
        RelationshipSummary(
            entity_id=entity_id,
            entity_type=EntityType.USER,  # Default, should be determined properly
        ),
    )

get_resolved_relationships async

get_resolved_relationships() -> list[Relationship]

Get resolved relationships (including those derived from group memberships).

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def get_resolved_relationships(self) -> list[Relationship]:
    """Get resolved relationships (including those derived from group memberships)."""
    resolved_relationships = []
    direct_relationships = await self.get_direct_relationships()

    # Add direct relationships
    resolved_relationships.extend(direct_relationships)

    # Resolve group memberships for access relationships
    group_relationships = [
        r
        for r in direct_relationships
        if r.from_entity_type == EntityType.GROUP and r.is_access_relationship
    ]

    for group_rel in group_relationships:
        group_members = await self._get_group_members(group_rel.from_entity_id)

        for member in group_members:
            resolved_rel = Relationship(
                from_entity_id=member.id,
                from_entity_type=(
                    EntityType.USER
                    if hasattr(member, "email")
                    else EntityType.GROUP
                ),
                to_entity_id=group_rel.to_entity_id,
                to_entity_type=group_rel.to_entity_type,
                relationship_type=group_rel.relationship_type,
                source_relationship_id=f"{group_rel.from_entity_id}_{group_rel.to_entity_id}",
            )
            resolved_relationships.append(resolved_rel)

    return resolved_relationships

invalidate_cache

invalidate_cache()

Invalidate the relationship cache to force refresh on next access.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def invalidate_cache(self):
    """Invalidate the relationship cache to force refresh on next access."""
    self._cache_valid = False

remove_relationship abstractmethod async

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

Remove a relationship.

Source code in src/crew_dcs/classes/subentity/Relationships.py

@abstractmethod
async def remove_relationship(
    self,
    target_entity_id: str,
    target_entity_type: EntityType,
    relationship_type: RelationshipType,
    **kwargs,
) -> bool:
    """Remove a relationship."""

DomoRelationshipManager

DomoRelationshipManager(auth: DomoAuth)

Main relationship manager that provides unified relationship management across all Domo objects.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def __init__(self, auth: DomoAuth):
    self.auth = auth
    self._controllers: dict[str, type] = {}

get_controller

get_controller(
    domo_object: DomoEntity,
) -> DomoRelationshipController | None

Get the appropriate relationship controller for a Domo object.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def get_controller(
    self, domo_object: DomoEntity
) -> DomoRelationshipController | None:
    """Get the appropriate relationship controller for a Domo object."""
    object_type = type(domo_object).__name__

    if object_type in self._controllers:
        return self._controllers[object_type](self.auth, domo_object)

    return None

get_entity_relationships async

get_entity_relationships(
    entity_id: str,
    entity_type: EntityType,
    relationship_types: (
        list[RelationshipType] | None
    ) = None,
) -> list[Relationship]

Get all relationships for an entity across all objects.

Source code in src/crew_dcs/classes/subentity/Relationships.py

async def get_entity_relationships(
    self,
    entity_id: str,
    entity_type: EntityType,
    relationship_types: list[RelationshipType] | None = None,
) -> list[Relationship]:
    """Get all relationships for an entity across all objects."""

register_controller

register_controller(
    object_type: str, controller_class: type
)

Register a relationship controller for a specific object type.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def register_controller(self, object_type: str, controller_class: type):
    """Register a relationship controller for a specific object type."""
    self._controllers[object_type] = controller_class

EntityType

Bases: DomoEnumMixin, Enum

Types of entities that can participate in relationships.

Relationship dataclass

Relationship(
    from_entity_id: str,
    from_entity_type: EntityType,
    to_entity_id: str,
    to_entity_type: EntityType,
    relationship_type: RelationshipType,
    created_date: str | None = None,
    created_by: str | None = None,
    effective_date: str | None = None,
    expiry_date: str | None = None,
    source_relationship_id: str | None = None,
    metadata: dict[str, Any] = dict(),
)

Bases: DomoBase

Represents a directional relationship between two entities.

is_access_relationship property

is_access_relationship: bool

Check if this is an access-related relationship.

is_lineage_relationship property

is_lineage_relationship: bool

Check if this is a lineage-related relationship.

is_membership_relationship property

is_membership_relationship: bool

Check if this is a membership-related relationship.

RelationshipSummary dataclass

RelationshipSummary(
    entity_id: str,
    entity_type: EntityType,
    relationships: list[Relationship] = list(),
)

Bases: DomoBase

Summary of relationships for a specific entity.

effective_access_level property

effective_access_level: RelationshipType

Get the highest access level from all access relationships.

get_access_relationships

get_access_relationships() -> list[Relationship]

Get all access-related relationships.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def get_access_relationships(self) -> list[Relationship]:
    """Get all access-related relationships."""
    return [r for r in self.relationships if r.is_access_relationship]

get_lineage_relationships

get_lineage_relationships() -> list[Relationship]

Get all lineage-related relationships.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def get_lineage_relationships(self) -> list[Relationship]:
    """Get all lineage-related relationships."""
    return [r for r in self.relationships if r.is_lineage_relationship]

get_membership_relationships

get_membership_relationships() -> list[Relationship]

Get all membership-related relationships.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def get_membership_relationships(self) -> list[Relationship]:
    """Get all membership-related relationships."""
    return [r for r in self.relationships if r.is_membership_relationship]

get_relationships_by_type

get_relationships_by_type(
    relationship_type: RelationshipType,
) -> list[Relationship]

Get all relationships of a specific type.

Source code in src/crew_dcs/classes/subentity/Relationships.py

def get_relationships_by_type(
    self, relationship_type: RelationshipType
) -> list[Relationship]:
    """Get all relationships of a specific type."""
    return [
        r for r in self.relationships if r.relationship_type == relationship_type
    ]

RelationshipType

Bases: DomoEnumMixin, Enum

Types of relationships between entities.