Skip to content

Condition

executor.engine.job.condition.Condition dataclass

Base class for condition

Source code in executor/engine/job/condition.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
@dataclass
class Condition():
    """Base class for condition"""

    def satisfy(self, engine: "Engine") -> bool:  # pragma: no cover
        """Check if the condition is satisfied."""
        return True

    def __or__(self, other: "Condition") -> "AnySatisfied":
        return AnySatisfied(conditions=[self, other])

    def __and__(self, other: "Condition") -> "AllSatisfied":
        return AllSatisfied(conditions=[self, other])

satisfy(engine)

Check if the condition is satisfied.

Source code in executor/engine/job/condition.py 
15
16
17
def satisfy(self, engine: "Engine") -> bool:  # pragma: no cover
    """Check if the condition is satisfied."""
    return True

executor.engine.job.condition.AfterAnother dataclass

Bases: Condition

Condition that the job is executed after another job is done/failed/cancelled.

Attributes:

Name Type Description
job_id str

The id of the job.
statuses Iterable[JobStatusType]

The statuses of the job that satisfy the condition.
Source code in executor/engine/job/condition.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@dataclass
class AfterAnother(Condition):
    """Condition that the job is executed after
    another job is done/failed/cancelled.

    Attributes:
        job_id: The id of the job.
        statuses: The statuses of the job that satisfy the condition.
    """

    job_id: str
    statuses: T.Iterable[JobStatusType] = ("done", "failed", "cancelled")

    def satisfy(self, engine) -> bool:
        try:
            another = engine.jobs.get_job_by_id(self.job_id)
        except Exception:
            return False
        if another.status in self.statuses:
            return True
        else:
            return False

executor.engine.job.condition.AfterOthers dataclass

Bases: Condition

Condition that the job is executed after other jobs are done/failed/cancelled.

Attributes:

Name Type Description
job_ids List[str]

The ids of the jobs.
statuses Iterable[JobStatusType]

The statuses of the jobs that satisfy the condition.
mode Literal['all', 'any']

The mode of the condition. If it is 'all', the job is executed after all other jobs are done/failed/cancelled. If it is 'any', the job is executed after any other job is done/failed/cancelled.
Source code in executor/engine/job/condition.py 
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
@dataclass
class AfterOthers(Condition):
    """Condition that the job is executed after
    other jobs are done/failed/cancelled.

    Attributes:
        job_ids: The ids of the jobs.
        statuses: The statuses of the jobs that satisfy the condition.
        mode: The mode of the condition. If it is 'all', the job is executed
            after all other jobs are done/failed/cancelled. If it is 'any',
            the job is executed after any other job is done/failed/cancelled.
    """

    job_ids: T.List[str]
    statuses: T.Iterable[JobStatusType] = ("done", "failed", "cancelled")
    mode: T.Literal['all', 'any'] = "all"

    def satisfy(self, engine) -> bool:
        other_job_satisfy = []
        for id_ in self.job_ids:
            try:
                job = engine.jobs.get_job_by_id(id_)
            except Exception:
                other_job_satisfy.append(False)
                continue
            s_ = job.status in self.statuses
            other_job_satisfy.append(s_)
        if self.mode == 'all':
            return all(other_job_satisfy)
        else:
            return any(other_job_satisfy)

executor.engine.job.condition.Combination dataclass

Bases: Condition

Base class for combination of conditions.

Attributes:

Name Type Description
conditions List[Condition]

The sub-conditions.
Source code in executor/engine/job/condition.py 
83
84
85
86
87
88
89
90
@dataclass
class Combination(Condition):
    """Base class for combination of conditions.

    Attributes:
        conditions: The sub-conditions.
    """
    conditions: T.List[Condition]

executor.engine.job.condition.AllSatisfied dataclass

Bases: Combination

Condition that the job is executed after all sub-conditions are satisfied.

Attributes:

Name Type Description
conditions List[Condition]

The sub-conditions.
Source code in executor/engine/job/condition.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@dataclass
class AllSatisfied(Combination):
    """Condition that the job is executed after all
    sub-conditions are satisfied.

    Attributes:
        conditions: The sub-conditions.
    """

    conditions: T.List[Condition]

    def satisfy(self, engine):
        """Check if the condition is satisfied."""
        return all([c.satisfy(engine) for c in self.conditions])

satisfy(engine)

Check if the condition is satisfied.

Source code in executor/engine/job/condition.py 
104
105
106
def satisfy(self, engine):
    """Check if the condition is satisfied."""
    return all([c.satisfy(engine) for c in self.conditions])

executor.engine.job.condition.AnySatisfied dataclass

Bases: Combination

Condition that the job is executed after any sub-condition is satisfied.

Attributes:

Name Type Description
conditions List[Condition]

The sub-conditions.
Source code in executor/engine/job/condition.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@dataclass
class AnySatisfied(Combination):
    """Condition that the job is executed after any
    sub-condition is satisfied.

    Attributes:
        conditions: The sub-conditions.
    """

    conditions: T.List[Condition]

    def satisfy(self, engine):
        """Check if the condition is satisfied."""
        return any([c.satisfy(engine) for c in self.conditions])

satisfy(engine)

Check if the condition is satisfied.

Source code in executor/engine/job/condition.py 
120
121
122
def satisfy(self, engine):
    """Check if the condition is satisfied."""
    return any([c.satisfy(engine) for c in self.conditions])

executor.engine.job.condition.TimeCondition dataclass

Bases: Condition

Source code in executor/engine/job/condition.py 
173
174
175
@dataclass
class TimeCondition(Condition):
    pass

executor.engine.job.condition.EveryPeriod dataclass

Bases: TimeCondition

Every period.

Attributes:

Name Type Description
period_str str

The period string. format: "1d", "1h", "1m", "1s"
last_submitted_at Optional[datetime]

The last submitted time.
immediate bool

Whether to submit the job immediately.
Source code in executor/engine/job/condition.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@dataclass
class EveryPeriod(TimeCondition):
    """Every period.

    Attributes:
        period_str: The period string.
            format: "1d", "1h", "1m", "1s"
        last_submitted_at: The last submitted time.
        immediate: Whether to submit the job immediately.
    """
    period_str: str
    last_submitted_at: T.Optional[datetime] = None
    immediate: bool = False

    def satisfy(self, _) -> bool:
        period = _parse_period_str(self.period_str)
        res: bool
        if self.last_submitted_at is None:
            self.last_submitted_at = datetime.now()
            res = self.immediate
        else:
            res = (datetime.now() - self.last_submitted_at >= period)
        if res:
            self.last_submitted_at = datetime.now()
        return res

executor.engine.job.condition.AfterClock dataclass

Bases: TimeCondition

After a specific clock.

Attributes:

Name Type Description
time_str str

The time string. format: "12:00", "12:00:00"
Source code in executor/engine/job/condition.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
@dataclass
class AfterClock(TimeCondition):
    """After a specific clock.

    Attributes:
        time_str: The time string.
            format: "12:00", "12:00:00"
    """
    time_str: str

    def satisfy(self, _) -> bool:
        hour, minute, second = _parse_clock_str(self.time_str)
        now = datetime.now()
        res = (
            (now.hour >= hour) &
            (now.minute >= minute) &
            (now.second >= second)
        )
        return res

executor.engine.job.condition.BeforeClock dataclass

Bases: TimeCondition

Before a specific clock.

Attributes:

Name Type Description
time_str str

The time string. format: "12:00", "12:00:00"
Source code in executor/engine/job/condition.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
@dataclass
class BeforeClock(TimeCondition):
    """Before a specific clock.

    Attributes:
        time_str: The time string.
            format: "12:00", "12:00:00"
    """
    time_str: str

    def satisfy(self, _) -> bool:
        hour, minute, second = _parse_clock_str(self.time_str)
        now = datetime.now()
        res = (
            (now.hour <= hour) &
            (now.minute <= minute) &
            (now.second <= second)
        )
        return res

executor.engine.job.condition.AfterWeekday dataclass

Bases: TimeCondition

After a specific weekday.

Attributes:

Name Type Description
weekday_str str

The weekday string. format: "monday", "mon", "tuesday", "tue", "wednesday", "wed", "thursday", "thu", "friday", "fri", "saturday", "sat", "sunday", "sun"
Source code in executor/engine/job/condition.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
@dataclass
class AfterWeekday(TimeCondition):
    """After a specific weekday.

    Attributes:
        weekday_str: The weekday string.
            format: "monday", "mon", "tuesday", "tue", "wednesday", "wed",
                "thursday", "thu", "friday", "fri", "saturday", "sat",
                "sunday", "sun"
    """
    weekday_str: str

    def satisfy(self, _) -> bool:
        weekday = _parse_weekday_str(self.weekday_str)
        now = datetime.now()
        res = (now.weekday() >= weekday)
        return res

executor.engine.job.condition.BeforeWeekday dataclass

Bases: TimeCondition

Before a specific weekday.

Attributes:

Name Type Description
weekday_str str

The weekday string. format: "monday", "mon", "tuesday", "tue", "wednesday", "wed", "thursday", "thu", "friday", "fri", "saturday", "sat", "sunday", "sun"
Source code in executor/engine/job/condition.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
@dataclass
class BeforeWeekday(TimeCondition):
    """Before a specific weekday.

    Attributes:
        weekday_str: The weekday string.
            format: "monday", "mon", "tuesday", "tue", "wednesday", "wed",
                "thursday", "thu", "friday", "fri", "saturday", "sat",
                "sunday", "sun"
    """
    weekday_str: str

    def satisfy(self, _) -> bool:
        weekday = _parse_weekday_str(self.weekday_str)
        now = datetime.now()
        res = (now.weekday() <= weekday)
        return res

executor.engine.job.condition.AfterTimepoint dataclass

Bases: TimeCondition

After a timepoint.

Attributes:

Name Type Description
timepoint datetime

The timepoint.
Source code in executor/engine/job/condition.py 
288
289
290
291
292
293
294
295
296
297
298
299
@dataclass
class AfterTimepoint(TimeCondition):
    """After a timepoint.

    Attributes:
        timepoint: The timepoint.
    """

    timepoint: datetime

    def satisfy(self, _) -> bool:
        return datetime.now() > self.timepoint

executor.engine.job.condition.BeforeTimepoint dataclass

Bases: TimeCondition

Before a timepoint.

Attributes:

Name Type Description
timepoint datetime

The timepoint.
Source code in executor/engine/job/condition.py 
302
303
304
305
306
307
308
309
310
311
312
@dataclass
class BeforeTimepoint(TimeCondition):
    """Before a timepoint.

    Attributes:
        timepoint: The timepoint.
    """
    timepoint: datetime

    def satisfy(self, _) -> bool:
        return datetime.now() < self.timepoint