get_dataset_from_catalog

kedro_dagster.utils.get_dataset_from_catalog(catalog, dataset_name)

Retrieve a dataset instance from a Kedro catalog across versions.

This helper avoids relying on the private _get_dataset when it is not available (as in some Kedro 1.x catalog wrappers) and falls back to mapping-like access when possible.

Parameters

Name	Type	Description	Default
catalog	CatalogProtocol	Kedro catalog or mapping-like object.	required
dataset_name	str	Name of the dataset to retrieve.	required

Returns

Type	Description
Any or None	Dataset instance if found, else None.

See Also

kedro_dagster.utils.is_nothing_asset_name : Uses this to check for sentinel Nothing datasets. kedro_dagster.catalog.CatalogTranslator : Translates catalog datasets into Dagster IO managers.

Source Code

View on GitHub

Show/Hide source

def get_dataset_from_catalog(catalog: "CatalogProtocol", dataset_name: str) -> Any | None:
    """Retrieve a dataset instance from a Kedro catalog across versions.

    This helper avoids relying on the private ``_get_dataset`` when it is not
    available (as in some Kedro 1.x catalog wrappers) and falls back to
    mapping-like access when possible.

    Parameters
    ----------
    catalog : CatalogProtocol
        Kedro catalog or mapping-like object.
    dataset_name : str
        Name of the dataset to retrieve.

    Returns
    -------
    Any or None
        Dataset instance if found, else ``None``.

    See Also
    --------
    `kedro_dagster.utils.is_nothing_asset_name` :
        Uses this to check for sentinel Nothing datasets.
    `kedro_dagster.catalog.CatalogTranslator` :
        Translates catalog datasets into Dagster IO managers.
    """
    result = None
    get_method = getattr(catalog, "_get_dataset", None)
    if callable(get_method):  # pragma: no cover
        try:
            result = get_method(dataset_name)
        except Exception:
            result = None
    else:
        # Mapping-like .get(name[, default])
        get_method = getattr(catalog, "get", None)
        if callable(get_method):
            try:
                result = get_method(dataset_name)
            except TypeError:  # pragma: no cover
                # Some get() signatures require a default
                try:
                    result = get_method(dataset_name, None)
                except Exception:
                    result = None
            except Exception:  # pragma: no cover
                result = None
        else:
            # Index access fallback
            try:
                result = catalog[dataset_name]
            except Exception:
                result = None

    if result is None:
        LOGGER.info(f"Dataset '{dataset_name}' not found in catalog.")

    return result