resolve_jobs¶

def resolve_jobs(
    config: KedroDagsterConfig, job_names: list[str], pipelines: Mapping[str, Any] | None = None
) -> dict[str, JobOptions]:
    """Resolve requested job names to ``JobOptions`` via literals + rendered bindings.

    Literal jobs are matched directly; any remaining name is looked up in the
    enumerated job set (see :func:`enumerate_jobs`). Raises a ``ValueError`` listing
    available names if a requested name is neither a literal job nor a rendered one.

    Notes
    -----
    Dagster's ``Definitions`` enumerates every job up front, so kedro-dagster has
    no single-job ``run -j <name>`` path; this function is retained for structural
    parity with the sibling ``kedro-azureml-pipeline`` plugin.
    """
    jobs = config.jobs or {}
    literals = {k: v for k, v in jobs.items() if not is_factory(k)}
    has_factories = any(is_factory(k) for k in jobs)

    resolved: dict[str, JobOptions] = {}
    all_jobs: dict[str, JobOptions] | None = None
    missing: list[str] = []
    for name in job_names:
        if name in literals:
            resolved[name] = literals[name]
            continue
        if not has_factories:  # nothing to render; only literal jobs exist
            missing.append(name)
            continue
        if all_jobs is None:
            all_jobs = enumerate_jobs(config, pipelines)
        if name in all_jobs:
            resolved[name] = all_jobs[name]
        else:
            missing.append(name)

    if missing:
        available = sorted(all_jobs if all_jobs is not None else literals)
        raise ValueError(
            f"Job(s) not found in config: {', '.join(sorted(missing))}. Available jobs: {', '.join(available)}"
        )
    return resolved