python-sdk/src/conductor/client/scheduler_client.py at 998bfb016982cd763ec4987212574955faa053c4 · conductor-oss/python-sdk · GitHub

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
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Optional, List
from conductor.client.http.models.workflow_schedule import WorkflowSchedule
from conductor.client.http.models.save_schedule_request import SaveScheduleRequest
from conductor.client.http.models.search_result_workflow_schedule_execution_model import \
    SearchResultWorkflowScheduleExecutionModel
from conductor.client.orkes.models.metadata_tag import MetadataTag


class SchedulerClient(ABC):
    @abstractmethod
    def save_schedule(self, save_schedule_request: SaveScheduleRequest):
        pass

    @abstractmethod
    def get_schedule(self, name: str) -> (Optional[WorkflowSchedule], str):
        pass

    @abstractmethod
    def get_all_schedules(self, workflow_name: Optional[str] = None) -> List[WorkflowSchedule]:
        pass

    @abstractmethod
    def get_next_few_schedule_execution_times(self,
                                              cron_expression: str,
                                              schedule_start_time: Optional[int] = None,
                                              schedule_end_time: Optional[int] = None,
                                              limit: Optional[int] = None,
                                              ) -> List[int]:
        pass

    @abstractmethod
    def delete_schedule(self, name: str):
        pass

    @abstractmethod
    def pause_schedule(self, name: str):
        pass

    @abstractmethod
    def pause_all_schedules(self):
        pass

    @abstractmethod
    def resume_schedule(self, name: str):
        pass

    @abstractmethod
    def resume_all_schedules(self):
        pass

    @abstractmethod
    def search_schedule_executions(self,
                                   start: Optional[int] = None,
                                   size: Optional[int] = None,
                                   sort: Optional[str] = None,
                                   free_text: Optional[str] = None,
                                   query: Optional[str] = None,
                                   ) -> SearchResultWorkflowScheduleExecutionModel:
        pass

    @abstractmethod
    def requeue_all_execution_records(self):
        pass

    @abstractmethod
    def set_scheduler_tags(self, tags: List[MetadataTag], name: str):
        pass

    @abstractmethod
    def get_scheduler_tags(self, name: str) -> List[MetadataTag]:
        pass

    @abstractmethod
    def delete_scheduler_tags(self, tags: List[MetadataTag], name: str) -> List[MetadataTag]:
        pass