dlt._workspace.deployment._trigger_helpers
parse_period_seconds
def parse_period_seconds(value: str) -> float
Parse a human period string (e.g. '5m', '1h', '30s') into seconds.
Also accepts bare numeric strings as seconds.
Raises:
ValueError- If the string cannot be parsed.
normalize_timeout
def normalize_timeout(
value: Union[int, float, str, TTimeoutSpec]) -> TTimeoutSpec
Normalize timeout input to TTimeoutSpec for the manifest.
parse_trigger
def parse_trigger(trigger: TTrigger) -> TParsedTrigger
Parse a normalized trigger string into a structured TParsedTrigger.
normalize_trigger
def normalize_trigger(trigger: Union[str, TTrigger]) -> TTrigger
Normalize a single user-provided trigger to canonical form.
Raises InvalidTrigger for manual: and pipeline_name: triggers which are
synthetic (added automatically from expose and deliver specs).
normalize_triggers
def normalize_triggers(
trigger: Union[None, str, TTrigger, Sequence[Union[str, TTrigger]],
Tuple[Union[str, TTrigger], ...]]
) -> List[TTrigger]
Normalize trigger input to a list of canonical TTrigger values.
is_selector
def is_selector(s: str) -> bool
Check if string looks like a trigger selector vs a bare job ref.
match_triggers_with_selectors
def match_triggers_with_selectors(job_type: str, triggers: List[TTrigger],
selectors: List[str]) -> List[TTrigger]
Return triggers that match any selector.
Job-type selectors (batch, interactive, stream, job) match ALL triggers.
pick_trigger
def pick_trigger(
matched: List[TTrigger],
default_trigger: Optional[TTrigger] = None) -> Optional[TTrigger]
Pick one trigger from a matched list. Prefer default_trigger if present.
maybe_parse_schedule
def maybe_parse_schedule(job_def: TJobDefinition) -> Optional[str]
Extract the cron expression from a job's schedule trigger.
Returns:
The cron expression string if the job has a schedule: trigger,
or None if no schedule trigger is found.
humanize_trigger
def humanize_trigger(trigger: TTrigger) -> str
Human-readable description of a trigger string for display in CLIs.