1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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
110
111
112
113
114
115
116
117
118
|
# ---------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ---------------------------------------------------------
# pylint: disable=unused-argument
from marshmallow import fields
from marshmallow.decorators import post_load
from azure.ai.ml._restclient.v2022_10_01_preview.models import ComputePowerAction, RecurrenceFrequency
from azure.ai.ml._restclient.v2022_10_01_preview.models import ScheduleStatus as ScheduleState
from azure.ai.ml._restclient.v2022_10_01_preview.models import TriggerType, WeekDay
from azure.ai.ml._schema.core.fields import NestedField, StringTransformedEnum, UnionField
from azure.ai.ml._schema.core.schema_meta import PatchedSchemaMeta
class BaseTriggerSchema(metaclass=PatchedSchemaMeta):
start_time = fields.Str()
time_zone = fields.Str()
class CronTriggerSchema(BaseTriggerSchema):
type = StringTransformedEnum(required=True, allowed_values=TriggerType.CRON)
expression = fields.Str(required=True)
@post_load
def make(self, data, **kwargs):
from azure.ai.ml.entities import CronTrigger
data.pop("type")
return CronTrigger(**data)
class RecurrenceScheduleSchema(metaclass=PatchedSchemaMeta):
week_days = fields.List(
StringTransformedEnum(
allowed_values=[
WeekDay.SUNDAY,
WeekDay.MONDAY,
WeekDay.TUESDAY,
WeekDay.WEDNESDAY,
WeekDay.THURSDAY,
WeekDay.FRIDAY,
WeekDay.SATURDAY,
],
)
)
hours = fields.List(fields.Int())
minutes = fields.List(fields.Int())
@post_load
def make(self, data, **kwargs):
from azure.ai.ml.entities import RecurrencePattern
return RecurrencePattern(**data)
class RecurrenceTriggerSchema(BaseTriggerSchema):
type = StringTransformedEnum(required=True, allowed_values=TriggerType.RECURRENCE)
frequency = StringTransformedEnum(
required=True,
allowed_values=[
RecurrenceFrequency.MINUTE,
RecurrenceFrequency.HOUR,
RecurrenceFrequency.DAY,
RecurrenceFrequency.WEEK,
RecurrenceFrequency.MONTH,
],
)
interval = fields.Int()
schedule = NestedField(RecurrenceScheduleSchema)
@post_load
def make(self, data, **kwargs):
from azure.ai.ml.entities import RecurrenceTrigger
data.pop("type")
return RecurrenceTrigger(**data)
class ComputeStartStopScheduleSchema(metaclass=PatchedSchemaMeta):
trigger = UnionField(
[
NestedField(CronTriggerSchema()),
NestedField(RecurrenceTriggerSchema()),
],
)
action = StringTransformedEnum(
required=True,
allowed_values=[
ComputePowerAction.START,
ComputePowerAction.STOP,
],
)
state = StringTransformedEnum(
allowed_values=[
ScheduleState.ENABLED,
ScheduleState.DISABLED,
],
)
schedule_id = fields.Str(dump_only=True)
provisioning_state = fields.Str(dump_only=True)
@post_load
def make(self, data, **kwargs):
from azure.ai.ml.entities import ComputeStartStopSchedule
return ComputeStartStopSchedule(**data)
class ComputeSchedulesSchema(metaclass=PatchedSchemaMeta):
compute_start_stop = fields.List(NestedField(ComputeStartStopScheduleSchema))
@post_load
def make(self, data, **kwargs):
from azure.ai.ml.entities import ComputeSchedules
return ComputeSchedules(**data)
|
