registry

registry

Lineage registry — decorators, registries, and import maps.

Extracted from base.py for separation of concerns. Contains: - PublisherResolution dataclass - FederatedLineageAuthRequiredError - Lineage type registry (register_lineage_type, get_lineage_type) - Lineage class registry (register_lineage, _try_register_lineage) - Import map for lazy registration

FederatedLineageAuthRequiredError

FederatedLineageAuthRequiredError(
    *,
    entity_id: str,
    entity_type: str,
    domo_instance: str | None,
    partial_lineage: list[DomoLineage_Link] | None = None
)

Bases: ClassError

Raised when federated lineage is requested without publisher auth.

.. deprecated:: This exception is no longer raised internally. Partial-lineage situations are now handled by logging a warning instead. Kept for backward-compatible except clauses in downstream code.

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

def __init__(
    self,
    *,
    entity_id: str,
    entity_type: str,
    domo_instance: str | None,
    partial_lineage: list[DomoLineage_Link] | None = None,
):
    self.partial_lineage = partial_lineage or []
    super().__init__(
        message=(
            "Federated lineage requires parent_auth or parent_auth_retrieval_fn to"
            " retrieve publisher-side dependencies."
        ),
        entity_id=entity_id,
        parent_class=entity_type,
        domo_instance=domo_instance,
    )

PublisherResolution dataclass

PublisherResolution(
    entity: Any | None = None,
    intermediate_entities: list[Any] = list(),
)

Result of resolving a publisher entity, with optional intermediate entities.

Attributes:

Name	Type	Description
entity	Any | None	The resolved publisher-side entity (e.g., a card or dataset), or None.
intermediate_entities	list[Any]	Entities between the publication and the resolved entity in the lineage chain (e.g., a page when a card is published indirectly as part of a page).

get_lineage_type

get_lineage_type(class_name: str) -> str

Get the lineage type string for a given class name.

Parameters:

Name	Type	Description	Default
class_name	str	The class name to look up	required

Returns:

Type	Description
str	The lineage type string (e.g., "DATAFLOW", "DATA_SOURCE")

Raises:

Type	Description
ValueError	If the class name is not registered

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

def get_lineage_type(class_name: str) -> str:
    """Get the lineage type string for a given class name.

    Args:
        class_name: The class name to look up

    Returns:
        The lineage type string (e.g., "DATAFLOW", "DATA_SOURCE")

    Raises:
        ValueError: If the class name is not registered
    """
    lineage_type = _LINEAGE_TYPE_REGISTRY.get(class_name)
    if lineage_type is None:
        raise ValueError(
            f"No lineage type registered for class '{class_name}'. "
            f"Known classes: {sorted(_LINEAGE_TYPE_REGISTRY.keys())}"
        )
    return lineage_type

register_lineage

register_lineage(*parent_class_names: str)

Decorator to register a DomoLineage subclass for parent class names.

Parameters:

Name	Type	Description	Default
*parent_class_names	str	One or more parent class names that should use this lineage class	()

Example

@register_lineage("DomoPage", "DomoPage_Default") @dataclass class DomoLineage_Page(DomoLineage): ...

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

def register_lineage(*parent_class_names: str):
    """Decorator to register a DomoLineage subclass for parent class names.

    Args:
        *parent_class_names: One or more parent class names that should use this lineage class

    Example:
        @register_lineage("DomoPage", "DomoPage_Default")
        @dataclass
        class DomoLineage_Page(DomoLineage):
            ...
    """

    def decorator(cls: type) -> type:
        for parent_class_name in parent_class_names:
            _LINEAGE_REGISTRY[parent_class_name] = cls
        return cls

    return decorator

register_lineage_type

register_lineage_type(*class_names: str, lineage_type: str)

Decorator to register lineage type for entity classes.

.. deprecated:: Direct use of @register_lineage_type is deprecated. Entity classes should set entity_type: ClassVar[str] on the class instead — init_subclass on DomoEntity auto-registers lineage types when has_lineage=True. This decorator is kept for backward compatibility and is called internally by the auto-registration system.

This decorator registers one or more entity class names with their corresponding lineage type string (e.g., "DATAFLOW", "DATA_SOURCE", "PAGE").

Parameters:

Name	Type	Description	Default
*class_names	str	One or more class names that should use this lineage type	()
lineage_type	str	The lineage type string (e.g., "DATAFLOW", "DATA_SOURCE")	required

Example

@register_lineage_type("DomoDataflow", lineage_type="DATAFLOW") @dataclass class DomoDataflow(DomoEntity_w_Lineage): ...

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

def register_lineage_type(*class_names: str, lineage_type: str):
    """Decorator to register lineage type for entity classes.

    .. deprecated::
        Direct use of @register_lineage_type is deprecated. Entity classes
        should set entity_type: ClassVar[str] on the class instead —
        __init_subclass__ on DomoEntity auto-registers lineage types when
        has_lineage=True. This decorator is kept for backward compatibility
        and is called internally by the auto-registration system.

    This decorator registers one or more entity class names with their
    corresponding lineage type string (e.g., "DATAFLOW", "DATA_SOURCE", "PAGE").

    Args:
        *class_names: One or more class names that should use this lineage type
        lineage_type: The lineage type string (e.g., "DATAFLOW", "DATA_SOURCE")

    Example:
        @register_lineage_type("DomoDataflow", lineage_type="DATAFLOW")
        @dataclass
        class DomoDataflow(DomoEntity_w_Lineage):
            ...
    """

    def decorator(cls):
        # Warn if called directly (not via __init_subclass__)
        # Internal calls from __init_subclass__ set _internal=True on the decorator
        if not getattr(decorator, "_internal", False):
            warnings.warn(
                f"@register_lineage_type is deprecated. "
                f"Set entity_type: ClassVar[str] = '{lineage_type}' on {cls.__name__} instead. "
                f"Lineage registration is now automatic via __init_subclass__.",
                DeprecationWarning,
                stacklevel=2,
            )
        for class_name in class_names:
            _LINEAGE_TYPE_REGISTRY[class_name] = lineage_type
        return cls

    return decorator