Skip to content

resolve_target

kedro_azureml_pipeline.factory.resolve_target(target, factories)

Render a single target into its (name, JobConfig), or None.

Selects the most-specific factory consistent with the target (see module docstring). Returns None if no factory is consistent with the target.

Source Code

Source code in src/kedro_azureml_pipeline/factory.py
def resolve_target(target: dict[str, str], factories: dict[str, JobConfig]) -> tuple[str, JobConfig] | None:
    """Render a single target into its ``(name, JobConfig)``, or ``None``.

    Selects the most-specific factory consistent with the target (see module
    docstring). Returns ``None`` if no factory is consistent with the target.
    """
    target_keys = set(target)
    job = target.get("job")

    candidates: list[tuple[str, JobConfig, str, int, int]] = []
    for key, config in factories.items():
        toks = _tokens(key)
        if not toks <= target_keys:
            continue
        name = _render_str(key, target)
        if "{" in name:  # unfilled placeholder remained
            continue
        if not _matches_job_type(name, job):
            continue
        candidates.append((key, config, name, _literal_len(key), len(toks)))

    if not candidates:
        return None

    # The canonical name is set by the most-general candidate (most tokens; ties
    # broken by most literal characters, then key) so it reflects every token of
    # the target (e.g. the full product, not a specific factory's literal prefix).
    canonical = max(candidates, key=lambda c: (c[4], c[3], c[0]))[2]
    applicable = [c for c in candidates if c[2] == canonical]
    # Most-specific applicable factory (most literal characters) supplies the body.
    key, config, name, *_ = max(applicable, key=lambda c: (c[3], c[0]))
    return name, _render_job(config, target)