get_match_pattern_from_catalog_resolver

kedro_dagster.utils.get_match_pattern_from_catalog_resolver(config_resolver, ds_name)

Return the matching dataset pattern from a CatalogConfigResolver.

Kedro 1.x exposes match_dataset_pattern while older Kedro versions exposed match_pattern. This helper abstracts over that difference and returns the first matching pattern string or None when no match is found or when the API is unavailable.

Parameters

Name	Type	Description	Default
config_resolver	CatalogConfigResolver	An instance of Kedro's CatalogConfigResolver.	required
ds_name	str	Dataset name to match against resolver patterns.	required

Returns

Type	Description
str or None	The first matching pattern or None.

See Also

kedro_dagster.utils.get_partition_mapping : Uses pattern matching to resolve partition mappings.

Source Code

View on GitHub

Show/Hide source

def get_match_pattern_from_catalog_resolver(config_resolver: "CatalogConfigResolver", ds_name: str) -> Any:
    """Return the matching dataset pattern from a CatalogConfigResolver.

    Kedro 1.x exposes ``match_dataset_pattern`` while older Kedro versions
    exposed ``match_pattern``. This helper abstracts over that difference and
    returns the first matching pattern string or ``None`` when no match is
    found or when the API is unavailable.

    Parameters
    ----------
    config_resolver : CatalogConfigResolver
        An instance of Kedro's CatalogConfigResolver.
    ds_name : str
        Dataset name to match against resolver patterns.

    Returns
    -------
    str or None
        The first matching pattern or ``None``.

    See Also
    --------
    `kedro_dagster.utils.get_partition_mapping` :
        Uses pattern matching to resolve partition mappings.
    """
    # Try both method names regardless of Kedro version, preferring the newer name
    for method_name in ("match_dataset_pattern", "match_pattern"):
        match_method = getattr(config_resolver, method_name, None)
        if callable(match_method):  # pragma: no cover
            try:
                return match_method(ds_name)
            except Exception:
                LOGGER.debug(f"Resolver.{method_name} failed for dataset '{ds_name}'", exc_info=True)
    return None