Skip to content

build_trigger

kedro_azureml_pipeline.scheduler.build_trigger(config)

Convert a ScheduleConfig into an Azure ML trigger object.

Parameters

Name Type Description Default
config ScheduleConfig

Schedule configuration with either a cron or recurrence definition.

 required

Returns

Type Description
CronTrigger or RecurrenceTrigger

Azure ML trigger ready for a JobSchedule.

See Also

resolve_schedule : Resolves config before building. build_job_schedule : Wraps the trigger into a schedule.

Source Code

View on GitHub

Show/Hide source 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def build_trigger(config: ScheduleConfig) -> CronTrigger | RecurrenceTrigger:
    """Convert a ``ScheduleConfig`` into an Azure ML trigger object.

    Parameters
    ----------
    config : ScheduleConfig
        Schedule configuration with either a ``cron`` or ``recurrence``
        definition.

    Returns
    -------
    CronTrigger or RecurrenceTrigger
        Azure ML trigger ready for a ``JobSchedule``.

    See Also
    --------
    [resolve_schedule][kedro_azureml_pipeline.scheduler.resolve_schedule] : Resolves config before building.
    [build_job_schedule][kedro_azureml_pipeline.scheduler.build_job_schedule] : Wraps the trigger into a schedule.
    """
    if config.cron:
        kwargs = {"expression": config.cron.expression, "time_zone": config.cron.time_zone}
        if config.cron.start_time:
            kwargs["start_time"] = config.cron.start_time
        if config.cron.end_time:
            kwargs["end_time"] = config.cron.end_time
        return CronTrigger(**kwargs)

    rec = config.recurrence
    assert rec is not None, "ScheduleConfig must have either cron or recurrence set"

    pattern: RecurrencePattern | None = None
    if rec.schedule:
        pattern_kwargs: dict[str, Any] = {}
        if rec.schedule.hours is not None:  # pragma: no branch - RecurrencePattern requires hours
            pattern_kwargs["hours"] = rec.schedule.hours
        if rec.schedule.minutes is not None:  # pragma: no branch - RecurrencePattern requires minutes
            pattern_kwargs["minutes"] = rec.schedule.minutes
        if rec.schedule.week_days is not None:
            pattern_kwargs["week_days"] = rec.schedule.week_days
        pattern = RecurrencePattern(**pattern_kwargs)

    return RecurrenceTrigger(
        frequency=rec.frequency,
        interval=rec.interval,
        time_zone=rec.time_zone,
        start_time=rec.start_time,
        end_time=rec.end_time,
        schedule=pattern,
    )