dlt._workspace.deployment._run_helpers

Shared run/serve orchestration helpers — pure manifest transforms, no CLI I/O.

promote_deployment_arg

def promote_deployment_arg(
        selector_or_job_ref: Optional[str],
        deployment: Optional[str]) -> Tuple[Optional[str], Optional[str]]

View source on GitHub

Promote a positional .py argument to the --deployment slot.

load_manifest_with_warnings

def load_manifest_with_warnings(
        name_or_path: str,
        *,
        use_all: bool = True
) -> Tuple[TJobsDeploymentManifest, str, List[str]]

View source on GitHub

Load a manifest, returning (manifest, manifest_hash, warnings).

resolve_selector

def resolve_selector(selector_or_job_ref: Optional[str],
                     manifest: TJobsDeploymentManifest,
                     *,
                     default_selector: str = "manual:") -> List[str]

View source on GitHub

Convert a CLI positional into a selector list. Bare refs become manual:<ref>.

select_candidates

def select_candidates(
        manifest: TJobsDeploymentManifest,
        selectors: List[str],
        *,
        forbidden_job_type: Optional[str] = None) -> List[TCandidate]

View source on GitHub

Match jobs against selectors, substituting manual: hits with the job's default trigger.

narrow_candidates

def narrow_candidates(candidates: List[TCandidate],
                      job_ref: Optional[str]) -> TCandidate

View source on GitHub

Pick one candidate; job_ref must match a candidate's ref exactly (no manifest-wide fallback).

select_single_job

def select_single_job(
        manifest: TJobsDeploymentManifest,
        selectors: List[str],
        *,
        forbidden_job_type: Optional[str] = None,
        job_ref: Optional[str] = None,
        available_selectors: Optional[List[str]] = None) -> TCandidate

View source on GitHub

Resolve selectors (+ optional job_ref) to exactly one matched job.

warn_missing_profiles

def warn_missing_profiles() -> List[str]

View source on GitHub

Advisory warnings when recommended profiles (prod, access) are missing locally.

resolve_refresh

def resolve_refresh(user_refresh: bool,
                    job_def: TJobDefinition) -> Tuple[bool, Optional[str]]

View source on GitHub

Apply a job's TRefreshPolicy to user_refresh. Returns (effective, warning_or_None).

resolve_profile

def resolve_profile(user_profile: Optional[str],
                    job_def: TJobDefinition) -> Tuple[str, Optional[str]]

View source on GitHub

Current profile (--profile wins over active); warns on declared-vs-active mismatch.

resolve_interval

def resolve_interval(user_start: Optional[str], user_end: Optional[str],
                     job_def: TJobDefinition, picked_trigger: TTrigger,
                     now_utc: datetime,
                     refresh: bool) -> Tuple[datetime, datetime, str]

View source on GitHub

Resolve (start_utc, end_utc, tz) for a run; user values win verbatim, otherwise clamp to declared interval.

build_runtime_entry_point

def build_runtime_entry_point(job_def: TJobDefinition, cli_config: Dict[str,
                                                                        str],
                              profile: str, refresh: bool,
                              interval_start: datetime, interval_end: datetime,
                              tz: str) -> TRuntimeEntryPoint

View source on GitHub

Assemble a TRuntimeEntryPoint from a job def and resolved context, without mutating job_def.

pick_launcher

def pick_launcher(entry_point: TRuntimeEntryPoint) -> str

View source on GitHub

Launcher module path: explicit override > function-based > module-level.

fetch_run_info

def fetch_run_info(
        *,
        selector: Optional[str] = None,
        selectors: Optional[List[str]] = None,
        deployment: Optional[str] = None,
        user_profile: Optional[str] = None,
        user_start: Optional[str] = None,
        user_end: Optional[str] = None,
        user_refresh: bool = False,
        cli_config: Optional[Dict[str, str]] = None,
        job_ref: Optional[str] = None,
        forbidden_job_type: Optional[str] = None,
        available_selectors: Optional[List[str]] = None,
        pick: Optional[TPickFn] = None,
        now_utc: Optional[datetime] = None) -> Optional[TRunJobInfo]

View source on GitHub

Resolve a run/serve request to a launchable TRunJobInfo.

Arguments:

• selector - User-supplied positional — a selector or a job ref.
• selectors - Pre-built selectors (e.g. ["pipeline_name:<name>"]). When set, selector is ignored.
• deployment - Path or module name of the deployment to load. Defaults to the workspace's default deployment module.
• user_profile - Profile override; wins over the active profile.
• user_start - ISO interval start override.
• user_end - ISO interval end override.
• user_refresh - Whether the user requested --refresh.
• cli_config - KEY=VALUE config overrides to merge into the entry point.
• job_ref - Narrow the matched candidates to this exact ref.
• forbidden_job_type - Skip jobs of this job_type (e.g. "interactive").
• available_selectors - Selectors used to scope NoMatchingJobs.available on no-match (e.g. ["batch"], ["interactive"]).
• pick - Callback invoked when more than one candidate matches and no job_ref was given.
• now_utc - Clock override for tests.

Returns:

A TRunJobInfo ready to launch, or None when the manifest has no jobs.

Raises:

• NoMatchingJobs - No job matched the selectors.
• AmbiguousJobSelector - Multiple jobs matched and pick is not provided.
• JobRefNotInCandidates - job_ref is not among matched candidates.