base

base

Domo entity system.

This package provides foundational classes for all Domo entities with support for authentication, relationships, lineage tracking, and entity management.

Modules: - base: Foundational classes and enhanced enums - entities: Core Domo entity classes and managers

The design provides a consistent interface across all Domo entity types while supporting advanced features like lineage tracking and relationships.

Access

Bases: DomoEnumMixin

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

DomoBase dataclass

DomoBase()

Bases: ABC

Abstract base class for all Domo objects.

This class serves as the foundation for all Domo entities and managers, providing a common interface and ensuring consistent implementation across the inheritance hierarchy.

Property Serialization Extension

Subclasses may declare a tuple __serialize_properties__ containing property (or attribute) names that should be appended to the default dataclass field serialization performed by :meth:to_dict.

Example::

from typing import ClassVar

@dataclass
class MyEntity(DomoBase):
    id: str
    value: int
    __serialize_properties__: ClassVar[tuple] = ("display_url",)

    @property
    def display_url(self) -> str:  # will be included automatically
        return f"https://example.com/{self.id}"

MyEntity(id="123", value=5).to_dict()
# {'id': '123', 'value': 5, 'displayUrl': 'https://example.com/123'}

MyEntity(id="123", value=5).to_dict(return_snake_case=True)
# {'id': '123', 'value': 5, 'display_url': 'https://example.com/123'}

Notes: * Only fields with repr=True and non-None values are emitted. * Properties listed in __serialize_properties__ are always included (even if None). * Properties that raise exceptions are skipped safely. * Dataclass fields take precedence over property names with the same identifier. * Use return_snake_case=True to get snake_case keys instead of camelCase.

to_dict

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

Convert dataclass to dictionary with camelCase or snake_case keys, excluding fields with repr=False.

Parameters:

Name	Type	Description	Default
override_fn	Callable | None	Optional callable that receives self and returns the final dictionary. Bypasses default behavior entirely.	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 values.

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

def to_dict(  # noqa: C901
    self, override_fn: Callable | None = None, return_snake_case: bool = False
) -> dict:
    """Convert dataclass to dictionary with camelCase or snake_case keys, excluding fields with repr=False.

    Args:
        override_fn: Optional callable that receives ``self`` and returns the final dictionary.
                    Bypasses default behavior entirely.
        return_snake_case: If True, return keys in snake_case. If False (default), return camelCase.

    Returns:
        dict: Dictionary with camelCase (default) or snake_case keys and corresponding values.
    """
    if override_fn:
        return override_fn(self)

    def _serialize_value(value: Any) -> Any:
        """Recursively serialize values, handling objects with to_dict methods."""
        if value is None:
            return None
        if hasattr(value, "to_dict") and callable(value.to_dict):
            return value.to_dict(return_snake_case=return_snake_case)
        if isinstance(value, list):
            return [_serialize_value(item) for item in value]
        if isinstance(value, dict):
            return {k: _serialize_value(v) for k, v in value.items()}
        return value

    # Start with dataclass fields (only include fields with repr=True)
    result: dict[str, Any] = {}

    # Build set of property names to force-include from __serialize_properties__
    force_include_fields = set()
    if getattr(self, "__serialize_properties__", None):
        force_include_fields = set(self.__serialize_properties__)  # type: ignore[attr-defined]

    for fld in fields(self):
        # Include field if: (repr=True and not None) OR (in __serialize_properties__)
        should_include = (fld.repr and getattr(self, fld.name) is not None) or (
            fld.name in force_include_fields
        )

        if should_include:
            key = (
                fld.name if return_snake_case else convert_snake_to_pascal(fld.name)
            )
            result[key] = _serialize_value(getattr(self, fld.name))

    # Append whitelisted properties / attributes (non-dataclass fields only)
    if getattr(self, "__serialize_properties__", None):
        for prop_name in self.__serialize_properties__:  # type: ignore[attr-defined]
            # Skip if it's already a dataclass field (already handled above)
            if any(f.name == prop_name for f in fields(self)):
                continue
            try:
                value = getattr(self, prop_name)
                # Include properties even if None to ensure consistent DataFrame columns
                key = (
                    prop_name
                    if return_snake_case
                    else convert_snake_to_pascal(prop_name)
                )
                result[key] = _serialize_value(value)
            except (
                AttributeError,
                TypeError,
                ValueError,
            ):  # pragma: no cover - defensive; skip failing properties
                continue

    return result

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().")

DomoEntityLineageProtocol

Bases: Protocol

Structural interface for lineage-capable entities.

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

DomoEnumMixin

Enhanced Enum mixin with case-insensitive lookup and default value support.

This mixin provides case-insensitive string matching and falls back to a default value when no match is found. All subclasses should define a 'default' member.

Example

class Status(DomoEnumMixin, Enum): ... ACTIVE = "active" ... INACTIVE = "inactive" ... default = "UNKNOWN" Status.get("ACTIVE") # Case insensitive Status.get("invalid")

get classmethod

get(value: Any) -> _EnumT | None

Get enum member by case-insensitive string lookup.

Parameters:

Name	Type	Description	Default
value	Any	String value to look up (case-insensitive)	required

Returns:

Type	Description
_EnumT | None	Enum member if found, otherwise the default member, or None if no default exists

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

@classmethod
def get(cls: type[_EnumT], value: Any) -> _EnumT | None:
    """Get enum member by case-insensitive string lookup.

    Args:
        value: String value to look up (case-insensitive)

    Returns:
        Enum member if found, otherwise the default member, or None if no default exists
    """
    if not isinstance(value, str):
        return getattr(cls, "default", None)

    # cls should be an Enum subclass at runtime
    for member in cls:  # type: ignore
        if member.name.lower() == value.lower():
            return member

    return getattr(cls, "default", None)

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.")

DomoPublicationProtocol

Bases: Protocol

Structural interface for publication objects.

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)

DomoSubscriptionProtocol

Bases: Protocol

Structural interface for subscription objects.

FederationContext dataclass

FederationContext(
    parent: DomoEntityLineageProtocol | None = None,
    subscription: DomoSubscriptionProtocol | None = None,
    parent_publication: (
        DomoPublicationProtocol | None
    ) = None,
    publisher_entity: (
        DomoEntityLineageProtocol | None
    ) = None,
    _parent_auth_fn: (
        Callable[[str], DomoAuth | Awaitable[DomoAuth]]
        | None
    ) = None,
    _parent_auth: DomoAuth | None = None,
    _content_type: str | None = None,
    _entity_id: str | None = None,
    _auth: DomoAuth | None = None,
)

Bases: DomoBase

Holds resolved federation state for federated entities.

Attached to subscriber-side entities that are copies from another instance. Manages subscription/publication discovery and caching.

Can be attached to a parent entity via constructor, or created standalone using from_entity_id() for probing publish state without a full entity.

auth property

auth: DomoAuth

Get auth from parent entity or standalone _auth.

entity_id property

entity_id: str

Get entity ID from stored value or parent.

check_if_published async

check_if_published(
    *,
    retrieve_parent_auth_fn: Callable[
        [str], DomoAuth | Awaitable[DomoAuth]
    ],
    entity_type: str,
    entity_id: str | None = None,
    context: RouteContext | None = None,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    max_subscriptions_to_check: int | None = None,
    **context_kwargs
) -> bool

Discover whether the entity participates in a subscription.

Source code in src/crew_dcs/classes/subentity/lineage/federation_context.py

async def check_if_published(
    self,
    *,
    retrieve_parent_auth_fn: Callable[[str], DomoAuth | Awaitable[DomoAuth]],
    entity_type: str,
    entity_id: str | None = None,
    context: RouteContext | None = None,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    max_subscriptions_to_check: int | None = None,
    **context_kwargs,
) -> bool:
    """Discover whether the entity participates in a subscription."""
    from .publish_resolver import PublishResolver

    if not retrieve_parent_auth_fn:
        raise ValueError(
            "retrieve_parent_auth_fn is required to resolve published entities."
        )

    # Build context from provided parameters
    context = RouteContext.build_context(
        context=context,
        session=session,
        debug_api=debug_api,
        **context_kwargs,
    )

    self._parent_auth_fn = retrieve_parent_auth_fn
    self._content_type = entity_type
    if entity_id:
        self._entity_id = entity_id

    target_id = self.entity_id

    await logger.debug(
        f"Checking if {entity_type} {target_id} is published",
        extra={
            "entity_type": entity_type,
            "entity_id": target_id,
            "subscriber_instance": self.auth.domo_instance,
        },
    )

    resolver = PublishResolver(
        subscriber_auth=self.auth,
        parent_auth_retrieval_fn=retrieve_parent_auth_fn,
        session=context.session,
        debug_api=context.debug_api,
        max_subscriptions_to_check=max_subscriptions_to_check,
    )

    try:
        # PublishResolver returns raw dict - hydrate into DomoSubscription
        from ...DomoEverywhere.core import DomoSubscription

        subscription_data = await resolver.get_subscription_for_entity(
            entity_type=entity_type,
            subscriber_entity_id=target_id,
        )

        self.subscription = DomoSubscription.from_dict(
            auth=self.auth,
            parent_publication=None,
            obj=subscription_data,
        )
        await logger.info(
            f"✅ {entity_type} {target_id} is published",
            extra={
                "entity_type": entity_type,
                "entity_id": target_id,
                "subscription_id": (
                    self.subscription.id if self.subscription else None
                ),
            },
        )
    except ValueError as exc:
        await logger.debug(
            f"❌ {entity_type} {target_id} is not published: {exc}",
            extra={
                "entity_type": entity_type,
                "entity_id": target_id,
                "error": str(exc),
            },
        )
        self.subscription = None

    return self.is_published

ensure_subscription async

ensure_subscription(
    *,
    retrieve_parent_auth_fn: (
        Callable[[str], DomoAuth | Awaitable[DomoAuth]]
        | None
    ) = None,
    parent_auth: DomoAuth | None = None,
    entity_type: str | None = None,
    entity_id: str | None = None,
    session: AsyncClient | None = None,
    debug_api: bool = False,
    max_subscriptions_to_check: int | None = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoSubscriptionProtocol | None

Ensure subscription data is loaded, running discovery if needed.

Parameters:

Name	Type	Description	Default
retrieve_parent_auth_fn	Callable[[str], DomoAuth | Awaitable[DomoAuth]] | None	Callable to retrieve parent auth by domain	None
parent_auth	DomoAuth | None	Pre-existing parent auth (alternative to retrieval function)	None
entity_type	str | None	Entity type (DATA_SOURCE, CARD, PAGE, etc.)	None
entity_id	str | None	The entity identifier	None
session	AsyncClient | None	HTTP session for API calls	None
debug_api	bool	Enable debug logging	False
max_subscriptions_to_check	int | None	Limit subscription search	None

Returns:

Type	Description
DomoSubscriptionProtocol | None	The subscription object if found

Raises:

Type	Description
ValueError	If no auth method available

Source code in src/crew_dcs/classes/subentity/lineage/federation_context.py

async def ensure_subscription(
    self,
    *,
    retrieve_parent_auth_fn: (
        Callable[[str], DomoAuth | Awaitable[DomoAuth]] | None
    ) = None,
    parent_auth: DomoAuth | None = None,
    entity_type: str | None = None,
    entity_id: str | None = None,
    session: httpx.AsyncClient | None = None,
    debug_api: bool = False,
    max_subscriptions_to_check: int | None = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoSubscriptionProtocol | None:
    """Ensure subscription data is loaded, running discovery if needed.

    Args:
        retrieve_parent_auth_fn: Callable to retrieve parent auth by domain
        parent_auth: Pre-existing parent auth (alternative to retrieval function)
        entity_type: Entity type (DATA_SOURCE, CARD, PAGE, etc.)
        entity_id: The entity identifier
        session: HTTP session for API calls
        debug_api: Enable debug logging
        max_subscriptions_to_check: Limit subscription search

    Returns:
        The subscription object if found

    Raises:
        ValueError: If no auth method available
    """

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

    if self.subscription:
        return self.subscription

    # Store parent_auth if provided
    if parent_auth:
        self._parent_auth = parent_auth

    fn = retrieve_parent_auth_fn or self._parent_auth_fn
    # Allow proceeding if we have either a retrieval function OR direct parent_auth
    if not fn and not self._parent_auth:
        raise ValueError(
            "retrieve_parent_auth_fn or parent_auth must be provided to resolve subscriptions."
        )

    # Determine content type - try stored, then parent if available
    content_type = entity_type or self._content_type
    if not content_type and self.parent is not None:
        content_type = self.parent.entity_type

    if not content_type:
        raise ValueError("entity_type must be provided or set via _content_type")

    # Use entity_id property which handles both standalone and parent modes
    target_id = entity_id or self.entity_id

    # If we have parent_auth but no fn, create a simple lambda that returns the auth
    effective_fn = fn
    if not effective_fn and self._parent_auth:
        effective_fn = lambda _domain, **_kwargs: self._parent_auth  # noqa: E731

    await self.check_if_published(
        retrieve_parent_auth_fn=effective_fn,
        entity_type=content_type,
        entity_id=target_id,
        context=context,
        session=session,
        debug_api=debug_api,
        max_subscriptions_to_check=max_subscriptions_to_check,
    )
    return self.subscription

from_entity_id classmethod

from_entity_id(
    *, auth: DomoAuth, entity_id: str, entity_type: str
) -> FederationContext

Create a federation context from entity identifiers (no parent entity required).

Use this factory when you need to check publish state without constructing a full entity object.

Parameters:

Name	Type	Description	Default
auth	DomoAuth	Authentication for the subscriber instance	required
entity_id	str	The entity identifier to check	required
entity_type	str	The entity type (DATA_SOURCE, CARD, PAGE, etc.)	required

Returns:

Type	Description
FederationContext	FederationContext instance ready for publish state checks

Source code in src/crew_dcs/classes/subentity/lineage/federation_context.py

@classmethod
def from_entity_id(
    cls,
    *,
    auth: DomoAuth,
    entity_id: str,
    entity_type: str,
) -> FederationContext:
    """Create a federation context from entity identifiers (no parent entity required).

    Use this factory when you need to check publish state without constructing
    a full entity object.

    Args:
        auth: Authentication for the subscriber instance
        entity_id: The entity identifier to check
        entity_type: The entity type (DATA_SOURCE, CARD, PAGE, etc.)

    Returns:
        FederationContext instance ready for publish state checks
    """
    return cls(
        parent=None,
        _auth=auth,
        _entity_id=str(entity_id),
        _content_type=entity_type,
    )

get_parent_publication async

get_parent_publication(
    *,
    parent_auth: DomoAuth | None = None,
    is_fetch_content_details: bool = True,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoPublicationProtocol

Fetch and cache the parent publication for this entity.

Parameters:

Name	Type	Description	Default
parent_auth	DomoAuth | None	Pre-existing publisher auth	None
is_fetch_content_details	bool	If True, automatically call get_content_details() on the publication before returning. This ensures content mapping is available for indirect resolution. Default: True.	True
context	RouteContext | None	Route context	None
**context_kwargs		Additional context args	{}

Returns:

Type	Description
DomoPublicationProtocol	DomoPublication with content details loaded (if is_fetch_content_details=True)

Source code in src/crew_dcs/classes/subentity/lineage/federation_context.py

async def get_parent_publication(
    self,
    *,
    parent_auth: DomoAuth | None = None,
    is_fetch_content_details: bool = True,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoPublicationProtocol:
    """Fetch and cache the parent publication for this entity.

    Args:
        parent_auth: Pre-existing publisher auth
        is_fetch_content_details: If True, automatically call get_content_details()
            on the publication before returning. This ensures content mapping is
            available for indirect resolution. Default: True.
        context: Route context
        **context_kwargs: Additional context args

    Returns:
        DomoPublication with content details loaded (if is_fetch_content_details=True)
    """
    context = RouteContext.build_context(
        context=context,
        **context_kwargs,
    )

    from ...DomoEverywhere.core import DomoPublication

    if not self.subscription:
        raise ValueError(
            "Subscription must be loaded before fetching parent publication."
        )

    publisher_auth = await self._resolve_parent_auth(parent_auth, context=context)

    if not self.parent_publication:
        self.parent_publication = await DomoPublication.get_by_id(
            publication_id=self.subscription.publication_id,
            auth=publisher_auth,
            context=context,
        )

    # Automatically fetch content details if requested
    if is_fetch_content_details and self.parent_publication:
        await self.parent_publication.get_content_details(
            subscriber_domain=self.auth.domo_instance,
            context=context,
        )

    return self.parent_publication

get_publisher_auth async

get_publisher_auth(
    parent_auth: Any = None,
    context: RouteContext | None = None,
    **context_kwargs
) -> DomoAuth

Public helper to resolve publisher authentication.

Source code in src/crew_dcs/classes/subentity/lineage/federation_context.py

async def get_publisher_auth(
    self,
    parent_auth: Any = None,
    context: RouteContext | None = None,
    **context_kwargs,
) -> DomoAuth:
    """Public helper to resolve publisher authentication."""

    context = RouteContext.build_context(
        context=context,
        **context_kwargs,
    )
    return await self._resolve_parent_auth(parent_auth, context=context)

hydrate_from_existing

hydrate_from_existing(
    *,
    subscription: DomoSubscriptionProtocol,
    parent_auth_retrieval_fn: (
        Callable[[str], DomoAuth | Awaitable[DomoAuth]]
        | None
    ) = None,
    parent_auth: DomoAuth | None = None,
    content_type: str,
    entity_id: str
)

Attach an already-discovered subscription to this helper.

Parameters:

Name	Type	Description	Default
subscription	DomoSubscriptionProtocol	The subscription object	required
parent_auth_retrieval_fn	Callable[[str], DomoAuth | Awaitable[DomoAuth]] | None	Callable to retrieve parent auth by domain	None
parent_auth	DomoAuth | None	Pre-existing parent auth (alternative to retrieval function)	None
content_type	str	Entity type (DATA_SOURCE, CARD, PAGE, etc.)	required
entity_id	str	The entity identifier	required

Source code in src/crew_dcs/classes/subentity/lineage/federation_context.py

def hydrate_from_existing(
    self,
    *,
    subscription: DomoSubscriptionProtocol,
    parent_auth_retrieval_fn: (
        Callable[[str], DomoAuth | Awaitable[DomoAuth]] | None
    ) = None,
    parent_auth: DomoAuth | None = None,
    content_type: str,
    entity_id: str,
):
    """Attach an already-discovered subscription to this helper.

    Args:
        subscription: The subscription object
        parent_auth_retrieval_fn: Callable to retrieve parent auth by domain
        parent_auth: Pre-existing parent auth (alternative to retrieval function)
        content_type: Entity type (DATA_SOURCE, CARD, PAGE, etc.)
        entity_id: The entity identifier
    """
    self.subscription = subscription
    self._parent_auth_fn = parent_auth_retrieval_fn
    self._parent_auth = parent_auth
    self._content_type = content_type
    self._entity_id = entity_id

FederationContextProtocol

Bases: Protocol

Structural interface for federation helpers.

PublishResolver dataclass

PublishResolver(
    subscriber_auth: DomoAuth,
    parent_auth_retrieval_fn: Callable[
        [str, RouteContext], Any | Awaitable[Any]
    ],
    session: AsyncClient | None = None,
    debug_api: bool = False,
    max_subscriptions_to_check: int | None = None,
)

Resolve subscriptions for subscriber-side entities.

get_subscription_for_entity async

get_subscription_for_entity(
    *,
    entity_type: ContentType,
    subscriber_entity_id: str,
    context: RouteContext | None = None,
    **context_kwargs: Any
) -> dict[str, Any]

Find the subscription that contains the given subscriber entity.

Returns:

Type	Description
dict[str, Any]	Raw subscription summary dict from the API. Callers should hydrate
dict[str, Any]	this into a DomoSubscription instance in the classes layer.

Source code in src/crew_dcs/classes/subentity/lineage/publish_resolver.py

async def get_subscription_for_entity(  # noqa: C901
    self,
    *,
    entity_type: ContentType,
    subscriber_entity_id: str,
    context: RouteContext | None = None,
    **context_kwargs: Any,
) -> dict[str, Any]:
    """Find the subscription that contains the given subscriber entity.

    Returns:
        Raw subscription summary dict from the API. Callers should hydrate
        this into a DomoSubscription instance in the classes layer.
    """
    if not self.parent_auth_retrieval_fn:
        raise ValueError(
            "parent_auth_retrieval_fn is required to resolve subscriptions. "
            "This function should accept a publisher_domain and return DomoAuth "
            "for that instance (sync or async)."
        )

    context = RouteContext.build_context(
        context=context,
        session=self.session,
        debug_api=self.debug_api,
        debug_num_stacks_to_drop=2,
        parent_class=self.__class__.__name__,
    )

    summaries_res = await publish_routes.get_subscription_summaries(
        auth=self.subscriber_auth,
        context=context,
    )

    if not summaries_res.is_success or not summaries_res.response:
        raise ValueError(
            f"Failed to retrieve subscriptions for instance "
            f"{self.subscriber_auth.domo_instance}"
        )

    subscriptions_checked = 0
    total_subscriptions = len(summaries_res.response)
    target_id = str(subscriber_entity_id)
    normalized_entity_type = (entity_type or "").upper()
    acceptable_content_types = {normalized_entity_type}
    # entity_type is now consistently "DATA_SOURCE" (not "DATASET"),
    # but the API may return either value, so accept both aliases
    if normalized_entity_type in ("DATA_SOURCE", "DATASET"):
        acceptable_content_types.update({"DATA_SOURCE", "DATASET"})

    await logger.debug(
        f"Found {total_subscriptions} subscriptions",
        extra={
            "entity_type": entity_type,
            "subscriber_entity_id": subscriber_entity_id,
            "subscriber_instance": self.subscriber_auth.domo_instance,
            "total_subscriptions": total_subscriptions,
        },
    )

    for summary in summaries_res.response:
        if (
            self.max_subscriptions_to_check is not None
            and subscriptions_checked >= self.max_subscriptions_to_check
        ):
            await logger.debug(
                f"Reached max subscriptions limit ({self.max_subscriptions_to_check}), stopping search",
                extra={"max_subscriptions": self.max_subscriptions_to_check},
            )
            break

        subscription_id = summary.get("subscriptionId")
        publication_id = summary.get("publicationId")
        subscriber_domain = summary.get("subscriberDomain")
        publisher_domain = summary.get("publisherDomain")

        if (
            not subscription_id
            or not publication_id
            or not subscriber_domain
            or not publisher_domain
        ):
            continue

        subscriptions_checked += 1

        sub_idx = (
            self.max_subscriptions_to_check
            if self.max_subscriptions_to_check
            else total_subscriptions
        )
        await logger.debug(
            f"Checking subscription {subscriptions_checked}/{sub_idx}: {subscription_id} - {publisher_domain}",
            extra={
                "subscription_id": subscription_id,
                "publisher_domain": publisher_domain,
                "publication_id": publication_id,
                "subscription_number": subscriptions_checked,
            },
        )

        publisher_auth = None

        try:
            publisher_auth = await self._get_publisher_auth(
                publisher_domain, context=context
            )
        except DomoError as exc:  # pragma: no cover - best effort logging
            await logger.warning(
                f"Unable to fetch auth for publisher {publisher_domain}: {exc}",
                extra={
                    "publisher_domain": publisher_domain,
                    "error": str(exc),
                    "error_type": type(exc).__name__,
                },
            )
            continue
        except (
            ValueError,
            KeyError,
            LookupError,
        ) as exc:  # pragma: no cover - auth lookup failure
            await logger.warning(
                f"Auth credentials not found for publisher {publisher_domain}: {exc}",
                extra={
                    "publisher_domain": publisher_domain,
                    "error": str(exc),
                    "error_type": type(exc).__name__,
                },
            )
            continue
        except (
            Exception  # noqa: BLE001
        ) as exc:  # pragma: no cover - unexpected error
            await logger.error(
                f"❌ Unexpected error fetching auth for publisher {publisher_domain}: {exc}",
                extra={
                    "publisher_domain": publisher_domain,
                    "error": str(exc),
                    "error_type": type(exc).__name__,
                },
                exc_info=True,
            )
            continue

        if not publisher_auth:
            continue

        try:
            content_res = await publish_routes.get_subscriber_content_details(
                auth=publisher_auth,
                publication_id=publication_id,
                subscriber_instance=subscriber_domain,
                context=context,
            )

        except DomoError as exc:  # pragma: no cover - best effort logging
            await logger.warning(
                f"Unable to fetch subscriber content details for {subscription_id}: {exc}",
                extra={
                    "subscription_id": subscription_id,
                    "error": str(exc),
                    "error_type": type(exc).__name__,
                },
            )
            continue
        except (
            Exception  # noqa: BLE001
        ) as exc:  # pragma: no cover - unexpected error
            await logger.error(
                f"❌ Unexpected error fetching subscriber content for {subscription_id}: {exc}",
                extra={
                    "subscription_id": subscription_id,
                    "error": str(exc),
                    "error_type": type(exc).__name__,
                },
                exc_info=True,
            )
            continue

        if not (content_res.is_success and content_res.response):
            await logger.debug(
                f"No subscriber content details for subscription {subscription_id}",
                extra={"subscription_id": subscription_id},
            )
            continue

        for item in content_res.response:
            content_type = (item.get("contentType") or "").upper()
            if (
                content_type in acceptable_content_types
                and str(item.get("subscriberObjectId")) == target_id
            ):
                await logger.info(
                    f"✅ Found {normalized_entity_type} {target_id} in subscription {subscription_id}",
                    extra={
                        "entity_type": normalized_entity_type,
                        "entity_id": target_id,
                        "subscription_id": subscription_id,
                        "publisher_domain": publisher_domain,
                    },
                )
                # Return raw dict - caller hydrates into DomoSubscription
                return summary

    # No subscription found
    if self.max_subscriptions_to_check:
        raise ValueError(
            f"Entity {subscriber_entity_id} (type {entity_type}) is not part of "
            f"any subscription after checking {subscriptions_checked} "
            f"subscriptions. Try increasing max_subscriptions_to_check "
            f"(currently {self.max_subscriptions_to_check})."
        )

    raise ValueError(
        f"Entity {subscriber_entity_id} (type {entity_type}) is not part of any "
        f"subscription after checking all {subscriptions_checked} subscriptions."
    )

Modules

• base
• entities
• exceptions