entities

entities

Access

Bases: DomoEnumMixin

abc for the concept of managing access levels to Domo entities.

DomoEntity dataclass

DomoEntity(auth: DomoAuth, id: str, raw: dict)

Bases: DomoBase

Base class for all Domo entities (datasets, cards, pages, users, etc.).

Provides core functionality including authentication, unique identification, data conversion utilities, and relationship management. All concrete entity types should inherit from this class or one of its subclasses.

Attributes:

Name	Type	Description
auth	DomoAuth	Authentication object for API requests (hidden in repr)
id	str	Unique identifier for the entity
raw	dict	Raw API response data for the entity (hidden in repr)
Relations	Any	Relationship controller for managing entity relationships

Class Attributes

entity_type: The Domo entity type string (e.g., "DATA_SOURCE", "PAGE"). Concrete subclasses MUST set this. Intermediate/abstract bases that don't override it are skipped by auto-registration. has_lineage: Whether this entity type supports lineage tracking. Defaults to False on DomoEntity, True on DomoEntity_w_Lineage.

Example

entity = SomeDomoEntity(auth=auth, id="123", raw={}) entity.display_url() # Implemented by subclass 'https://mycompany.domo.com/...'

display_url abstractmethod property

display_url: str

Generate the URL to display this entity in the Domo interface.

This method should return the direct URL to view the entity in Domo's web interface, allowing users to navigate directly to the entity.

Returns:

Name	Type	Description
str	str	Complete URL to view the entity in Domo

Raises:

Type	Description
NotImplementedError	Must be implemented by subclasses

entity_name property

entity_name: str

Get the display name for this entity.

Tries common name fields in order: name, title, display_name. Falls back to entity ID if no name is found.

Subclasses can override this property to use entity-specific name fields.

Returns:

Type	Description
str	Display name for the entity, or entity ID as fallback

create async classmethod

create(auth: DomoAuth, **kwargs) -> DomoEntity

Create a new entity and return it (classmethod: no instance yet).

Creatable entities override this to build the create payload (typically via a builder) and call the corresponding route.

Raises:

Type	Description
NotImplementedError	If this entity type does not support creation.

Source code in src/crew_dcs/base/entities.py

@classmethod
async def create(cls, auth: DomoAuth, **kwargs) -> DomoEntity:
    """Create a new entity and return it (classmethod: no instance yet).

    Creatable entities override this to build the create payload (typically
    via a builder) and call the corresponding route.

    Raises:
        NotImplementedError: If this entity type does not support creation.
    """
    raise NotImplementedError(f"{cls.__name__} does not implement create().")

delete async

delete(**kwargs)

Delete this entity.

Raises:

Type	Description
NotImplementedError	If this entity type does not support deletion.

Source code in src/crew_dcs/base/entities.py

async def delete(self, **kwargs):
    """Delete this entity.

    Raises:
        NotImplementedError: If this entity type does not support deletion.
    """
    raise NotImplementedError(f"{type(self).__name__} does not implement delete().")

dispatch_get_by_id async classmethod

dispatch_get_by_id(
    auth: DomoAuth,
    entity_type: str,
    entity_id: str,
    **kwargs
)

Route entity_type + entity_id to the right subclass.get_entity_by_id().

The entity class must already be imported (which triggers init_subclass registration). If you get a "not registered" error, import the entity module before calling this method.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
entity_type	str	The Domo entity type string (e.g., "DATA_SOURCE", "PAGE")	required
entity_id	str	Unique identifier of the entity to retrieve	required
**kwargs		Additional arguments passed to the subclass's get_entity_by_id	{}

Returns:

Type	Description
	An instance of the appropriate DomoEntity subclass

Raises:

Type	Description
ValueError	If entity_type is not registered

Source code in src/crew_dcs/base/entities.py

@classmethod
async def dispatch_get_by_id(
    cls,
    auth: DomoAuth,
    entity_type: str,
    entity_id: str,
    **kwargs,
):
    """Route entity_type + entity_id to the right subclass.get_entity_by_id().

    The entity class must already be imported (which triggers __init_subclass__
    registration). If you get a "not registered" error, import the entity module
    before calling this method.

    Args:
        auth: Authentication object for API requests
        entity_type: The Domo entity type string (e.g., "DATA_SOURCE", "PAGE")
        entity_id: Unique identifier of the entity to retrieve
        **kwargs: Additional arguments passed to the subclass's get_entity_by_id

    Returns:
        An instance of the appropriate DomoEntity subclass

    Raises:
        ValueError: If entity_type is not registered
    """
    target = cls._entity_dispatch_registry.get(entity_type)
    if not target:
        raise ValueError(
            f"No class registered for entity_type='{entity_type}'. "
            f"Registered types: {sorted(cls._entity_dispatch_registry.keys())}. "
            f"Ensure the entity module has been imported."
        )
    return await target.get_entity_by_id(auth=auth, entity_id=entity_id, **kwargs)

from_dict abstractmethod classmethod

from_dict(auth: DomoAuth, obj: dict[str, Any])

Create an entity instance from a dictionary representation.

This method should be implemented by subclasses to handle the conversion from API response dictionaries to entity objects.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
obj	dict[str, Any]	Dictionary representation of the entity from the API	required

Raises:

Type	Description
NotImplementedError	Must be implemented by subclasses

Source code in src/crew_dcs/base/entities.py

@classmethod
@abc.abstractmethod
def from_dict(cls, auth: DomoAuth, obj: dict[str, Any]):
    """Create an entity instance from a dictionary representation.

    This method should be implemented by subclasses to handle the conversion
    from API response dictionaries to entity objects.

    Args:
        auth: Authentication object for API requests
        obj: Dictionary representation of the entity from the API

    Raises:
        NotImplementedError: Must be implemented by subclasses
    """
    raise NotImplementedError("This method should be implemented by subclasses.")

get_by_id abstractmethod async classmethod

get_by_id(
    auth: DomoAuth,
    id: str,
    debug_num_stacks_to_drop=2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
)

Fetch an entity by its unique identifier.

This method should be implemented by subclasses to handle entity-specific retrieval logic from the Domo API.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
entity_id	str	Unique identifier of the entity to retrieve	required

Raises:

Type	Description
NotImplementedError	Must be implemented by subclasses

Source code in src/crew_dcs/base/entities.py

@classmethod
@abc.abstractmethod
async def get_by_id(
    cls,
    auth: DomoAuth,
    id: str,
    debug_num_stacks_to_drop=2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
):
    """Fetch an entity by its unique identifier.

    This method should be implemented by subclasses to handle entity-specific
    retrieval logic from the Domo API.

    Args:
        auth (DomoAuth): Authentication object for API requests
        entity_id (str): Unique identifier of the entity to retrieve

    Raises:
        NotImplementedError: Must be implemented by subclasses
    """
    raise NotImplementedError("This method should be implemented by subclasses.")

get_entity_by_id abstractmethod async classmethod

get_entity_by_id(
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Any | None = None,
    parent_auth: Any | None = None,
    **kwargs
)

Fetch an entity by its ID

This method should be implemented by subclasses to fetch the specific entity type while ensuring lineage tracking is properly initialized.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication object for API requests	required
entity_id	str	Unique identifier of the entity to retrieve	required
debug_num_stacks_to_drop	int	Number of stack frames to drop for debug logging (default: 2)	2
debug_api	bool	Enable API debug logging (default: False)	False
session	AsyncClient | None	Optional HTTP client session	None
check_if_published	bool | None	When True attempt to resolve publish subscriptions. If None and either parent_auth or parent_auth_retrieval_fn is provided, subclasses should default to True.	None
parent_auth_retrieval_fn	Any | None	Callable used to obtain publisher auth for publish checks	None
parent_auth	Any | None	Pre-existing publisher auth (alternative to parent_auth_retrieval_fn)	None
**kwargs		Additional arguments passed to the underlying get_by_id method	{}

Raises:

Type	Description
NotImplementedError	Must be implemented by subclasses

Source code in src/crew_dcs/base/entities.py

@classmethod
@abc.abstractmethod
async def get_entity_by_id(
    cls,
    auth: DomoAuth,
    entity_id: str,
    debug_num_stacks_to_drop: int = 2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    check_if_published: bool | None = None,
    parent_auth_retrieval_fn: Any | None = None,
    parent_auth: Any | None = None,
    **kwargs,
):
    """Fetch an entity by its ID

    This method should be implemented by subclasses to fetch the specific
    entity type while ensuring lineage tracking is properly initialized.

    Args:
        auth (DomoAuth): Authentication object for API requests
        entity_id (str): Unique identifier of the entity to retrieve
        debug_num_stacks_to_drop (int): Number of stack frames to drop for debug logging (default: 2)
        debug_api (bool): Enable API debug logging (default: False)
        session (httpx.AsyncClient | None): Optional HTTP client session
        check_if_published (bool | None): When True attempt to resolve publish subscriptions.
            If None and either parent_auth or parent_auth_retrieval_fn is provided, subclasses should default to True.
        parent_auth_retrieval_fn: Callable used to obtain publisher auth for publish checks
        parent_auth: Pre-existing publisher auth (alternative to parent_auth_retrieval_fn)
        **kwargs: Additional arguments passed to the underlying get_by_id method

    Raises:
        NotImplementedError: Must be implemented by subclasses
    """
    raise NotImplementedError("This method should be implemented by subclasses.")

refresh async

refresh(
    debug_num_stacks_to_drop=2,
    debug_api: bool = False,
    session: AsyncClient | None = None,
    **kwargs
)

Refresh this instance from the API using its id and auth.

Source code in src/crew_dcs/base/entities.py

@log_call(level_name="class", log_level="DEBUG", color="cyan")
async def refresh(
    self,
    debug_num_stacks_to_drop=2,
    debug_api: bool = False,
    session: httpx.AsyncClient | None = None,
    **kwargs,
):
    """Refresh this instance from the API using its id and auth."""

    try:
        await logger.debug(
            f"Refreshing {self.__class__.__name__} - {self.id} in {self.auth.domo_instance}..."
        )
        result = await type(self).get_entity_by_id(
            auth=self.auth,
            entity_id=self.id,
            debug_num_stacks_to_drop=debug_num_stacks_to_drop,
            debug_api=debug_api,
            session=session,
            **kwargs,
        )
    except DomoError as e:
        await logger.error(
            f"Failed to refresh {self.__class__.__name__} - {self.id} in {self.auth.domo_instance}: {e}"
        )
        raise

    # Spread attributes from result to self
    if isinstance(result, type(self)):
        self.__dict__.update(
            {k: v for k, v in result.__dict__.items() if v is not None}
        )
    return self

to_dict

to_dict(
    override_fn: Callable | None = None,
    return_snake_case: bool = False,
) -> dict

Convert all dataclass attributes to a dictionary in camelCase or snake_case.

This method is useful for serializing entity data for API requests or data export operations.

Only fields with repr=True are included, plus any properties listed in serialize_properties.

Parameters:

Name	Type	Description	Default
override_fn	Callable | None	Custom conversion function to override default behavior	None
return_snake_case	bool	If True, return keys in snake_case. If False (default), return camelCase.	False

Returns:

Name	Type	Description
dict	dict	Dictionary with camelCase (default) or snake_case keys and corresponding attribute values

Example

entity.to_dict() {'id': '123', 'displayName': 'My Entity', 'displayUrl': '...', ...} entity.to_dict(return_snake_case=True) {'id': '123', 'display_name': 'My Entity', 'display_url': '...', ...}

Source code in src/crew_dcs/base/entities.py

def to_dict(
    self, override_fn: Callable | None = None, return_snake_case: bool = False
) -> dict:
    """Convert all dataclass attributes to a dictionary in camelCase or snake_case.

    This method is useful for serializing entity data for API requests
    or data export operations.

    Only fields with repr=True are included, plus any properties listed in
    __serialize_properties__.

    Args:
        override_fn (Callable | None): Custom conversion function to override default behavior
        return_snake_case (bool): If True, return keys in snake_case. If False (default), return camelCase.

    Returns:
        dict: Dictionary with camelCase (default) or snake_case keys and corresponding attribute values

    Example:
        >>> entity.to_dict()
        {'id': '123', 'displayName': 'My Entity', 'displayUrl': '...', ...}
        >>> entity.to_dict(return_snake_case=True)
        {'id': '123', 'display_name': 'My Entity', 'display_url': '...', ...}
    """

    # Use parent's implementation which handles repr filtering and __serialize_properties__
    return super().to_dict(
        override_fn=override_fn, return_snake_case=return_snake_case
    )

update async

update(**kwargs) -> DomoEntity

Update this entity in place and return self.

Raises:

Type	Description
NotImplementedError	If this entity type does not support update.

Source code in src/crew_dcs/base/entities.py

async def update(self, **kwargs) -> DomoEntity:
    """Update this entity in place and return ``self``.

    Raises:
        NotImplementedError: If this entity type does not support update.
    """
    raise NotImplementedError(f"{type(self).__name__} does not implement update().")

DomoEntity_w_Lineage dataclass

DomoEntity_w_Lineage(auth: DomoAuth, id: str, raw: dict)

Bases: DomoEntity

Entity with lineage tracking capabilities.

Extends DomoEntity to include lineage tracking functionality, enabling entities to track their relationships and dependencies within the Domo ecosystem.

Attributes:

Name	Type	Description
Lineage	Any | None	Lineage tracking object for dependency management (hidden in repr)
Federation	Any | None	Federation context for publish/subscribe state (hidden in repr)
__skip_lineage_registration__	Any | None	Class attribute to opt out of registration requirement. Set to True for abstract or intermediate base classes that should not be registered.

name property

name: str

Get the display name for this entity.

All entities with lineage provide a name property for consistent identification in lineage diagrams and reports.

Entities that use 'title' (DomoCard, DomoPage) should override this as: @property def name(self) -> str: return self.title or f"Untitled {self.entity_type}"

Entities with a 'name' attribute (DomoDataset, DomoDataflow, DomoPublication) will use their dataclass field directly.

Returns:

Type	Description
str	Display name for the entity

Raises:

Type	Description
AttributeError	If entity doesn't have name, title, or override this property

enable_federation_support

enable_federation_support()

Ensure this entity has an attached FederationContext helper. returns FederationContext instance.

Source code in src/crew_dcs/base/entities.py

def enable_federation_support(self):
    """Ensure this entity has an attached FederationContext helper.
    returns FederationContext instance.
    """
    if self.Federation is None:
        from ..classes.subentity.lineage.federation_context import (
            FederationContext,
        )

        self.Federation = FederationContext(parent=self)
    return self.Federation

probe_is_published async classmethod

probe_is_published(
    entity_id: str,
    subscriber_auth: DomoAuth,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    max_subscriptions_to_check: int | None = None,
    context: RouteContext | None = None,
)

Check if entity is published (federated) - works for all entity types.

Uses lineage type registry to determine entity type automatically.

Parameters:

Name	Type	Description	Default
entity_id	str	Entity identifier to check	required
subscriber_auth	DomoAuth	Auth for subscriber instance	required
parent_auth	DomoAuth | None	Optional pre-existing publisher auth	None
parent_auth_retrieval_fn	Callable | None	Callable to retrieve publisher auth	None
session	AsyncClient | None	Optional HTTP session for reuse	None
debug_api	bool	Enable debug logging	False
max_subscriptions_to_check	int | None	Limit subscription checking	None
context	RouteContext | None	Optional pre-built context	None

Returns:

Type	Description
	FederationContext with subscription info if published

Raises:

Type	Description
ValueError	If class not registered with lineage type or missing auth

Source code in src/crew_dcs/base/entities.py

@classmethod
async def probe_is_published(
    cls,
    entity_id: str,
    subscriber_auth: DomoAuth,
    parent_auth: DomoAuth | None = None,
    parent_auth_retrieval_fn: Callable | None = None,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    max_subscriptions_to_check: int | None = None,
    context: RouteContext | None = None,
):
    """Check if entity is published (federated) - works for all entity types.

    Uses lineage type registry to determine entity type automatically.

    Args:
        entity_id: Entity identifier to check
        subscriber_auth: Auth for subscriber instance
        parent_auth: Optional pre-existing publisher auth
        parent_auth_retrieval_fn: Callable to retrieve publisher auth
        session: Optional HTTP session for reuse
        debug_api: Enable debug logging
        max_subscriptions_to_check: Limit subscription checking
        context: Optional pre-built context

    Returns:
        FederationContext with subscription info if published

    Raises:
        ValueError: If class not registered with lineage type or missing auth
    """
    from ..classes.subentity.lineage.federation_context import FederationContext

    # Use entity_type ClassVar (set by __init_subclass__ registration)
    entity_type = getattr(cls, "entity_type", "")
    if not entity_type:
        raise ValueError(
            f"Cannot determine entity type for {cls.__name__}. "
            f"Ensure the class sets entity_type: ClassVar[str]."
        )

    context = RouteContext.build_context(
        context=context, session=session, debug_api=debug_api
    )

    if not parent_auth_retrieval_fn and not parent_auth:
        raise ValueError(
            f"parent_auth_retrieval_fn is required to determine publish state for {entity_type}."
        )

    # If parent_auth is provided but not parent_auth_retrieval_fn, create a simple
    # retrieval function that returns the provided auth for any domain
    effective_retrieval_fn = parent_auth_retrieval_fn
    if parent_auth and not parent_auth_retrieval_fn:

        def effective_retrieval_fn(domain, **_kwargs):
            return parent_auth

    await logger.debug(
        f"Probing if {entity_type} {entity_id} is published",
        extra={
            "entity_type": entity_type,
            "entity_id": entity_id,
            "class_name": cls.__name__,
        },
    )

    probe = FederationContext.from_entity_id(
        auth=subscriber_auth,
        entity_id=str(entity_id),
        entity_type=entity_type,
    )

    is_published = await probe.check_if_published(
        retrieve_parent_auth_fn=effective_retrieval_fn,
        entity_type=entity_type,
        context=context,
        max_subscriptions_to_check=max_subscriptions_to_check,
    )

    await logger.debug(
        f"Probe result: {entity_type} {entity_id} is_published={is_published}",
        extra={
            "entity_type": entity_type,
            "entity_id": entity_id,
            "is_published": is_published,
            "has_subscription": probe.subscription is not None,
        },
    )

    return probe

DomoManager dataclass

DomoManager(auth: DomoAuth)

Bases: DomoBase

Base class for entity managers that handle collections of entities.

Provides the foundation for manager classes that handle operations on collections of entities (e.g., DatasetManager, CardManager).

Attributes:

Name	Type	Description
auth	DomoAuth	Authentication object for API requests (hidden in repr)

get abstractmethod async

get(*args: Any, **kwargs: Any) -> list[DomoEntity]

Retrieve entities based on provided criteria.

Must be implemented by subclasses to handle entity-specific retrieval and filtering logic.

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments for entity retrieval	()
**kwargs	Any	Keyword arguments for filtering and options	{}

Returns:

Type	Description
list[DomoEntity]	list[DomoEntity]: List of entity instances retrieved

Raises:

Type	Description
NotImplementedError	Must be implemented by subclasses

Source code in src/crew_dcs/base/entities.py

@abc.abstractmethod
async def get(self, *args: Any, **kwargs: Any) -> list[DomoEntity]:
    """Retrieve entities based on provided criteria.

    Must be implemented by subclasses to handle entity-specific
    retrieval and filtering logic.

    Args:
        *args: Positional arguments for entity retrieval
        **kwargs: Keyword arguments for filtering and options

    Returns:
        list[DomoEntity]: List of entity instances retrieved

    Raises:
        NotImplementedError: Must be implemented by subclasses
    """
    raise NotImplementedError("This method should be implemented by subclasses.")

DomoSubEntity dataclass

DomoSubEntity(parent: DomoEntity)

Bases: DomoBase

Base class for entities that belong to a parent entity.

Handles entities that are sub-components of other entities, such as columns in a dataset or slides in a page. Automatically inherits authentication and parent references.

Attributes:

Name	Type	Description
parent	DomoEntity	Reference to the parent entity
auth		Authentication object (inherited from parent, hidden in repr)

from_parent classmethod

from_parent(parent: DomoEntity)

Create a sub-entity instance from a parent entity.

Parameters:

Name	Type	Description	Default
parent	DomoEntity	The parent entity to derive from	required

Returns:

Name	Type	Description
DomoSubEntity		New sub-entity instance with inherited properties

Source code in src/crew_dcs/base/entities.py

@classmethod
def from_parent(cls, parent: DomoEntity):
    """Create a sub-entity instance from a parent entity.

    Args:
        parent (DomoEntity): The parent entity to derive from

    Returns:
        DomoSubEntity: New sub-entity instance with inherited properties
    """
    return cls(parent=parent)