|
1 | 1 | from __future__ import annotations |
| 2 | +import logging |
2 | 3 | from typing import Optional, List |
3 | 4 |
|
4 | 5 | from conductor.client.configuration.configuration import Configuration |
|
11 | 12 | from conductor.client.orkes.orkes_base_client import OrkesBaseClient |
12 | 13 | from conductor.client.scheduler_client import SchedulerClient |
13 | 14 |
|
| 15 | +logger = logging.getLogger(__name__) |
| 16 | + |
14 | 17 |
|
15 | 18 | class OrkesSchedulerClient(OrkesBaseClient, SchedulerClient): |
16 | 19 | def __init__(self, configuration: Configuration): |
@@ -133,7 +136,107 @@ def get_scheduler_tags(self, name: str) -> List[MetadataTag]: |
133 | 136 | def delete_scheduler_tags(self, tags: List[MetadataTag], name: str) -> List[MetadataTag]: |
134 | 137 | self.schedulerResourceApi.delete_tag_for_schedule(tags, name) |
135 | 138 |
|
136 | | - def _start_workflow(self, request) -> str: |
137 | | - # Enables SchedulerClient.run_now — OrkesBaseClient already carries the |
138 | | - # workflow resource API. |
139 | | - return self.workflowResourceApi.start_workflow(body=request) |
| 139 | + # ── schedule lifecycle (domain surface) ────────────────────────── |
| 140 | + # |
| 141 | + # Built over the endpoint methods above. Reads/writes/lists have NO domain |
| 142 | + # twins by design — get_schedule/save_schedule/get_all_schedules are the |
| 143 | + # source of truth. These methods raise the typed errors from |
| 144 | + # conductor.client.ai.schedule_errors (ScheduleNotFound, |
| 145 | + # InvalidCronExpression, ...) instead of raw ApiException. |
| 146 | + # The conductor.client.ai imports are deliberately lazy: this module is on |
| 147 | + # the OrkesClients import path and must not pull the agent surface at |
| 148 | + # import time. |
| 149 | + |
| 150 | + def pause(self, wire_name: str, reason: Optional[str] = None) -> None: |
| 151 | + """Pause one schedule by wire name (typed errors; ``reason`` is OSS-only).""" |
| 152 | + from conductor.client.ai.schedule import _translate |
| 153 | + |
| 154 | + try: |
| 155 | + if reason is None: |
| 156 | + self.pause_schedule(wire_name) |
| 157 | + else: |
| 158 | + self.pause_schedule(wire_name, reason=reason) |
| 159 | + except Exception as exc: # noqa: BLE001 |
| 160 | + raise _translate(exc) from exc |
| 161 | + |
| 162 | + def resume(self, wire_name: str) -> None: |
| 163 | + """Resume one schedule by wire name (typed errors).""" |
| 164 | + from conductor.client.ai.schedule import _translate |
| 165 | + |
| 166 | + try: |
| 167 | + self.resume_schedule(wire_name) |
| 168 | + except Exception as exc: # noqa: BLE001 |
| 169 | + raise _translate(exc) from exc |
| 170 | + |
| 171 | + def delete(self, wire_name: str) -> None: |
| 172 | + """Delete one schedule by wire name (typed errors).""" |
| 173 | + from conductor.client.ai.schedule import _translate |
| 174 | + |
| 175 | + try: |
| 176 | + self.delete_schedule(wire_name) |
| 177 | + except Exception as exc: # noqa: BLE001 |
| 178 | + raise _translate(exc) from exc |
| 179 | + |
| 180 | + def preview_next( |
| 181 | + self, cron: str, n: int = 5, start_at: Optional[int] = None, end_at: Optional[int] = None |
| 182 | + ) -> List[int]: |
| 183 | + """Next ``n`` epoch-ms fire times for ``cron`` (typed errors).""" |
| 184 | + from conductor.client.ai.schedule import _translate |
| 185 | + |
| 186 | + try: |
| 187 | + times = self.get_next_few_schedule_execution_times( |
| 188 | + cron_expression=cron, |
| 189 | + schedule_start_time=start_at, |
| 190 | + schedule_end_time=end_at, |
| 191 | + limit=n, |
| 192 | + ) |
| 193 | + return list(times) if times else [] |
| 194 | + except Exception as exc: # noqa: BLE001 |
| 195 | + raise _translate(exc) from exc |
| 196 | + |
| 197 | + def run_now(self, info: "ScheduleInfo") -> str: # noqa: F821 |
| 198 | + """Fire the schedule's agent once with its stored input. Returns execution id.""" |
| 199 | + from conductor.client.ai.schedule import _translate |
| 200 | + from conductor.client.http.models.start_workflow_request import StartWorkflowRequest |
| 201 | + |
| 202 | + req = StartWorkflowRequest(name=info.agent, input=dict(info.input)) |
| 203 | + try: |
| 204 | + return self.workflowResourceApi.start_workflow(body=req) |
| 205 | + except Exception as exc: # noqa: BLE001 |
| 206 | + raise _translate(exc) from exc |
| 207 | + |
| 208 | + def reconcile(self, agent_name: str, desired: Optional[List["Schedule"]]) -> None: # noqa: F821 |
| 209 | + """Apply the declarative tri-state semantics from spec §5.1: |
| 210 | +
|
| 211 | + - ``desired is None``: no-op |
| 212 | + - ``desired == []``: delete every schedule whose workflow == agent_name |
| 213 | + - ``desired == [...]``: upsert listed, delete others scoped to this agent |
| 214 | + """ |
| 215 | + from conductor.client.ai.schedule import ( |
| 216 | + _check_unique_names, |
| 217 | + _list_infos, |
| 218 | + _prefix, |
| 219 | + _to_save_request, |
| 220 | + _translate, |
| 221 | + ) |
| 222 | + |
| 223 | + if desired is None: |
| 224 | + return |
| 225 | + _check_unique_names(desired) |
| 226 | + existing = _list_infos(self, agent_name) |
| 227 | + existing_wire_by_short = {info.short_name: info.name for info in existing} |
| 228 | + desired_short = {s.name for s in desired} |
| 229 | + |
| 230 | + for short, wire in existing_wire_by_short.items(): |
| 231 | + if short not in desired_short: |
| 232 | + logger.info("Pruning schedule %s for agent %s", wire, agent_name) |
| 233 | + self.delete(wire) |
| 234 | + |
| 235 | + for s in desired: |
| 236 | + logger.info( |
| 237 | + "Upserting schedule %s for agent %s", _prefix(agent_name, s.name), agent_name |
| 238 | + ) |
| 239 | + try: |
| 240 | + self.save_schedule(_to_save_request(s, agent_name)) |
| 241 | + except Exception as exc: # noqa: BLE001 |
| 242 | + raise _translate(exc) from exc |
0 commit comments