Bases: BaseModel
The scheduling configuration. It can be triggerd based on several ways:
Interval: the sync will be trigged based on certain interval(minutes/hours/days/weeks)
Cron: the sync will be trigged based on cron expression https://en.wikipedia.org/wiki/Cron.
Visual: the sync will be trigged based a visual cron configuration on UI
DBT-cloud: the sync will be trigged based on a dbt cloud job
Attributes:
schedule (Union[CronSchedule, DBTSchedule, IntervalSchedule, VisualCronSchedule]):
type (str):
Source code in prefect_hightouch/api_client/models/sync_schedule.py
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134 | class SyncSchedule(BaseModel):
"""The scheduling configuration. It can be triggerd based on several ways:
Interval: the sync will be trigged based on certain interval(minutes/hours/days/weeks)
Cron: the sync will be trigged based on cron expression https://en.wikipedia.org/wiki/Cron.
Visual: the sync will be trigged based a visual cron configuration on UI
DBT-cloud: the sync will be trigged based on a dbt cloud job
Attributes:
schedule (Union[CronSchedule, DBTSchedule, IntervalSchedule, VisualCronSchedule]):
type (str):
"""
schedule: Union[
CronSchedule, DBTSchedule, IntervalSchedule, VisualCronSchedule
] = None
type: str = None
additional_properties: Dict[str, Any] = Field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
schedule: Dict[str, Any]
if isinstance(self.schedule, IntervalSchedule):
schedule = self.schedule.to_dict()
elif isinstance(self.schedule, CronSchedule):
schedule = self.schedule.to_dict()
elif isinstance(self.schedule, VisualCronSchedule):
schedule = self.schedule.to_dict()
else:
schedule = self.schedule.to_dict()
type = self.type
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"schedule": schedule,
"type": type,
}
)
return field_dict
@classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
if src_dict is None or src_dict is UNSET:
return {}
d = {k: v if v is not None else UNSET for k, v in src_dict.items()}
def _parse_schedule(
data: object,
) -> Union[CronSchedule, DBTSchedule, IntervalSchedule, VisualCronSchedule]:
try:
if not isinstance(data, dict):
raise TypeError()
schedule_type_0 = IntervalSchedule.from_dict(data)
return schedule_type_0
except: # noqa: E722
pass
try:
if not isinstance(data, dict):
raise TypeError()
schedule_type_1 = CronSchedule.from_dict(data)
return schedule_type_1
except: # noqa: E722
pass
try:
if not isinstance(data, dict):
raise TypeError()
schedule_type_2 = VisualCronSchedule.from_dict(data)
return schedule_type_2
except: # noqa: E722
pass
if not isinstance(data, dict):
raise TypeError()
schedule_type_3 = DBTSchedule.from_dict(data)
return schedule_type_3
schedule = _parse_schedule(d.pop("schedule"))
type = d.pop("type")
sync_schedule = cls(
schedule=schedule,
type=type,
)
sync_schedule.additional_properties = d
return sync_schedule
@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())
def __getitem__(self, key: str) -> Any:
return self.additional_properties.get(key)
def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
def __contains__(self, key: str) -> bool:
return key in self.additional_properties
|
additional_keys: List[str]
property
additional_properties: Dict[str, Any] = Field(default_factory=dict)
class-attribute
instance-attribute
schedule: Union[CronSchedule, DBTSchedule, IntervalSchedule, VisualCronSchedule] = None
class-attribute
instance-attribute
type: str = None
class-attribute
instance-attribute
from_dict
classmethod
Source code in prefect_hightouch/api_client/models/sync_schedule.py
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 | @classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
if src_dict is None or src_dict is UNSET:
return {}
d = {k: v if v is not None else UNSET for k, v in src_dict.items()}
def _parse_schedule(
data: object,
) -> Union[CronSchedule, DBTSchedule, IntervalSchedule, VisualCronSchedule]:
try:
if not isinstance(data, dict):
raise TypeError()
schedule_type_0 = IntervalSchedule.from_dict(data)
return schedule_type_0
except: # noqa: E722
pass
try:
if not isinstance(data, dict):
raise TypeError()
schedule_type_1 = CronSchedule.from_dict(data)
return schedule_type_1
except: # noqa: E722
pass
try:
if not isinstance(data, dict):
raise TypeError()
schedule_type_2 = VisualCronSchedule.from_dict(data)
return schedule_type_2
except: # noqa: E722
pass
if not isinstance(data, dict):
raise TypeError()
schedule_type_3 = DBTSchedule.from_dict(data)
return schedule_type_3
schedule = _parse_schedule(d.pop("schedule"))
type = d.pop("type")
sync_schedule = cls(
schedule=schedule,
type=type,
)
sync_schedule.additional_properties = d
return sync_schedule
|
to_dict
Source code in prefect_hightouch/api_client/models/sync_schedule.py
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 | def to_dict(self) -> Dict[str, Any]:
schedule: Dict[str, Any]
if isinstance(self.schedule, IntervalSchedule):
schedule = self.schedule.to_dict()
elif isinstance(self.schedule, CronSchedule):
schedule = self.schedule.to_dict()
elif isinstance(self.schedule, VisualCronSchedule):
schedule = self.schedule.to_dict()
else:
schedule = self.schedule.to_dict()
type = self.type
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"schedule": schedule,
"type": type,
}
)
return field_dict
|