Skip to content

Commit 652e58b

Browse files
authored
Schedules hotfix for DTS dashboard (#169)
* Schedules hotfix for DTS dashboard
1 parent 30499fe commit 652e58b

4 files changed

Lines changed: 340 additions & 34 deletions

File tree

CHANGELOG.md

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,10 @@ CHANGED
1717

1818
- Changed the default large-payload externalization threshold (`LargePayloadStorageOptions.threshold_bytes`) from 900,000 bytes to 262,144 bytes (256 KiB), matching the .NET SDK default. Behavioral change (not source/binary breaking): payloads larger than 256 KiB are now externalized by default.
1919

20+
FIXED
21+
22+
- Fixed schedules created with `durabletask.scheduled` not appearing in the Durable Task Scheduler dashboard. The schedule entity state is now persisted in a format compatible with the .NET SDK: the `status` is serialized as its numeric enum value, the `interval` as a .NET `TimeSpan` string, and the `orchestration_input` as a JSON-encoded string, all using .NET property names, so the dashboard can read it without a JSON deserialization error.
23+
2024
## v1.7.0
2125

2226
ADDED

durabletask/scheduled/models.py

Lines changed: 180 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,15 @@
88

99
from durabletask.internal.helpers import ensure_aware
1010
from durabletask.scheduled.schedule_status import ScheduleStatus
11+
from durabletask.serialization import JsonDataConverter
1112

1213
MINIMUM_INTERVAL = timedelta(seconds=1)
1314

15+
# Serializer used to (de)serialize the orchestration input to/from a JSON string
16+
# for persistence. Matches the .NET SDK, which stores ``OrchestrationInput`` as a
17+
# string so the Durable Task Scheduler dashboard can read the raw entity state.
18+
_INPUT_CONVERTER = JsonDataConverter()
19+
1420

1521
def _validate_interval(interval: timedelta) -> timedelta:
1622
if interval <= timedelta(0):
@@ -25,7 +31,15 @@ def _to_iso(value: datetime | None) -> str | None:
2531

2632

2733
def _from_iso(value: str | None) -> datetime | None:
28-
return datetime.fromisoformat(value) if value else None
34+
if not value:
35+
return None
36+
# .NET serializes ``DateTimeOffset`` with a numeric offset, but tolerate a
37+
# trailing ``Z`` so states written by other producers still parse on the
38+
# Python versions that predate ``fromisoformat`` accepting ``Z``. Only the
39+
# trailing designator is normalized so an interior ``Z`` is left untouched.
40+
if value.endswith("Z"):
41+
value = value[:-1] + "+00:00"
42+
return datetime.fromisoformat(value)
2943

3044

3145
def _interval_to_seconds(value: timedelta | None) -> float | None:
@@ -36,6 +50,119 @@ def _interval_from_seconds(value: float | None) -> timedelta | None:
3650
return timedelta(seconds=value) if value is not None else None
3751

3852

53+
# Number of 100-nanosecond ticks per second, matching the .NET ``TimeSpan``
54+
# resolution used when formatting the fractional component.
55+
_TICKS_PER_SECOND = 10_000_000
56+
57+
58+
def _interval_to_timespan(value: timedelta | None) -> str | None:
59+
"""Format a ``timedelta`` as a .NET ``TimeSpan`` (``[-][d.]hh:mm:ss[.fffffff]``).
60+
61+
The Durable Task Scheduler dashboard deserializes the schedule interval into
62+
a .NET ``TimeSpan``, whose JSON converter only accepts this constant format.
63+
"""
64+
if value is None:
65+
return None
66+
negative = value < timedelta(0)
67+
value = abs(value)
68+
days = value.days
69+
hours, remainder = divmod(value.seconds, 3600)
70+
minutes, seconds = divmod(remainder, 60)
71+
if days:
72+
formatted = f"{days}.{hours:02d}:{minutes:02d}:{seconds:02d}"
73+
else:
74+
formatted = f"{hours:02d}:{minutes:02d}:{seconds:02d}"
75+
if value.microseconds:
76+
ticks = value.microseconds * 10 # microseconds -> 100-ns ticks
77+
formatted += f".{ticks:07d}"
78+
return f"-{formatted}" if negative else formatted
79+
80+
81+
def _interval_from_timespan(value: str) -> timedelta:
82+
"""Parse a .NET ``TimeSpan`` string (``[-][d.]hh:mm:ss[.fffffff]``)."""
83+
text = value.strip()
84+
negative = text.startswith("-")
85+
if negative:
86+
text = text[1:]
87+
88+
fraction = 0.0
89+
if "." in text:
90+
head, _, tail = text.rpartition(".")
91+
# A dot before the first ``:`` is the day separator, not a fraction.
92+
if ":" in tail:
93+
# e.g. ``1.02:03:04`` -- the ``.`` separates days from the clock.
94+
days_part, _, clock = text.partition(".")
95+
days = int(days_part)
96+
hours, minutes, seconds = (int(p) for p in clock.split(":"))
97+
else:
98+
# ``head`` holds ``[d.]hh:mm:ss`` and ``tail`` the ticks fraction.
99+
fraction = int(tail.ljust(7, "0")[:7]) / _TICKS_PER_SECOND
100+
days, hours, minutes, seconds = _split_clock(head)
101+
else:
102+
days, hours, minutes, seconds = _split_clock(text)
103+
104+
result = timedelta(days=days, hours=hours, minutes=minutes,
105+
seconds=seconds) + timedelta(seconds=fraction)
106+
return -result if negative else result
107+
108+
109+
def _split_clock(text: str) -> tuple[int, int, int, int]:
110+
"""Split ``[d.]hh:mm:ss`` into ``(days, hours, minutes, seconds)``."""
111+
days = 0
112+
if "." in text:
113+
days_part, _, text = text.partition(".")
114+
days = int(days_part)
115+
hours, minutes, seconds = (int(part) for part in text.split(":"))
116+
return days, hours, minutes, seconds
117+
118+
119+
def _get(data: dict[str, Any], *keys: str, default: Any = None) -> Any:
120+
"""Return the first present key from ``data``.
121+
122+
Reads tolerate both the .NET-compatible PascalCase keys and the legacy
123+
snake_case keys written by earlier Python workers.
124+
"""
125+
for key in keys:
126+
if key in data:
127+
return data[key]
128+
return default
129+
130+
131+
def _parse_interval(data: dict[str, Any]) -> timedelta:
132+
"""Read the interval from either the .NET ``Interval`` or legacy field."""
133+
timespan = _get(data, "Interval", "interval")
134+
if isinstance(timespan, str):
135+
return _interval_from_timespan(timespan)
136+
seconds = _get(data, "interval_seconds")
137+
if seconds is not None:
138+
return timedelta(seconds=seconds)
139+
raise KeyError("interval")
140+
141+
142+
def _encode_orchestration_input(value: Any) -> str | None:
143+
"""Serialize the orchestration input to a JSON string for persistence.
144+
145+
The Durable Task Scheduler dashboard (and the .NET SDK) model the persisted
146+
``OrchestrationInput`` as a string, so the raw input object is serialized to
147+
a JSON string here and parsed back by :func:`_decode_orchestration_input`.
148+
"""
149+
if value is None:
150+
return None
151+
return _INPUT_CONVERTER.serialize(value)
152+
153+
154+
def _decode_orchestration_input(value: Any) -> Any:
155+
"""Reconstruct the raw orchestration input from its persisted JSON string."""
156+
if value is None or not isinstance(value, str):
157+
return value
158+
try:
159+
return _INPUT_CONVERTER.deserialize(value)
160+
except (ValueError, TypeError):
161+
# Not a JSON document (an unexpected shape); return it unchanged rather
162+
# than failing to load the schedule.
163+
return value
164+
165+
39166
@dataclass
40167
class ScheduleCreationOptions:
41168
"""Options for creating a new schedule."""
@@ -230,29 +357,39 @@ def _validate(self):
230357
raise ValueError("start_at cannot be later than end_at.")
231358

232359
def to_json(self) -> dict[str, Any]:
360+
# Serialized with .NET-compatible property names and value shapes so the
361+
# Durable Task Scheduler dashboard can deserialize the raw entity state:
362+
# PascalCase keys and the interval as a .NET ``TimeSpan`` string.
233363
return {
234-
"schedule_id": self.schedule_id,
235-
"orchestration_name": self.orchestration_name,
236-
"interval_seconds": self.interval.total_seconds(),
237-
"orchestration_input": self.orchestration_input,
238-
"orchestration_instance_id": self.orchestration_instance_id,
239-
"start_at": _to_iso(self.start_at),
240-
"end_at": _to_iso(self.end_at),
241-
"start_immediately_if_late": self.start_immediately_if_late,
364+
"ScheduleId": self.schedule_id,
365+
"OrchestrationName": self.orchestration_name,
366+
"Interval": _interval_to_timespan(self.interval),
367+
"OrchestrationInput": _encode_orchestration_input(self.orchestration_input),
368+
"OrchestrationInstanceId": self.orchestration_instance_id,
369+
"StartAt": _to_iso(self.start_at),
370+
"EndAt": _to_iso(self.end_at),
371+
"StartImmediatelyIfLate": self.start_immediately_if_late,
242372
}
243373

244374
@classmethod
245375
def from_json(cls, data: dict[str, Any]) -> "ScheduleConfiguration":
246376
config = cls(
247-
data["schedule_id"],
248-
data["orchestration_name"],
249-
timedelta(seconds=data["interval_seconds"]),
377+
_get(data, "ScheduleId", "schedule_id"),
378+
_get(data, "OrchestrationName", "orchestration_name"),
379+
_parse_interval(data),
250380
)
251-
config.orchestration_input = data.get("orchestration_input")
252-
config.orchestration_instance_id = data.get("orchestration_instance_id")
253-
config.start_at = _from_iso(data.get("start_at"))
254-
config.end_at = _from_iso(data.get("end_at"))
255-
config.start_immediately_if_late = bool(data.get("start_immediately_if_late", False))
381+
if "OrchestrationInput" in data:
382+
# New .NET-compatible states store the input as a JSON string.
383+
config.orchestration_input = _decode_orchestration_input(data["OrchestrationInput"])
384+
else:
385+
# Legacy states stored the raw (already-parsed) input object.
386+
config.orchestration_input = data.get("orchestration_input")
387+
config.orchestration_instance_id = _get(
388+
data, "OrchestrationInstanceId", "orchestration_instance_id")
389+
config.start_at = _from_iso(_get(data, "StartAt", "start_at"))
390+
config.end_at = _from_iso(_get(data, "EndAt", "end_at"))
391+
config.start_immediately_if_late = bool(
392+
_get(data, "StartImmediatelyIfLate", "start_immediately_if_late", default=False))
256393
return config
257394

258395

@@ -273,16 +410,18 @@ def refresh_execution_token(self):
273410

274411
def to_json(self) -> dict[str, Any]:
275412
# ``schedule_configuration`` is returned as the object itself; the
276-
# serializer recurses into it and fires its own ``to_json`` hook. Only
277-
# this type's non-JSON-native leaves (datetimes) are converted here.
413+
# serializer recurses into it and fires its own ``to_json`` hook. Keys
414+
# and value shapes mirror the .NET ``ScheduleState`` so the Durable Task
415+
# Scheduler dashboard can deserialize the raw entity state: PascalCase
416+
# names, the status as its numeric ordinal, and datetimes as ISO strings.
278417
return {
279-
"status": self.status.value,
280-
"execution_token": self.execution_token,
281-
"last_run_at": _to_iso(self.last_run_at),
282-
"next_run_at": _to_iso(self.next_run_at),
283-
"schedule_created_at": _to_iso(self.schedule_created_at),
284-
"schedule_last_modified_at": _to_iso(self.schedule_last_modified_at),
285-
"schedule_configuration": self.schedule_configuration,
418+
"Status": self.status.to_dotnet_ordinal(),
419+
"ExecutionToken": self.execution_token,
420+
"LastRunAt": _to_iso(self.last_run_at),
421+
"NextRunAt": _to_iso(self.next_run_at),
422+
"ScheduleCreatedAt": _to_iso(self.schedule_created_at),
423+
"ScheduleLastModifiedAt": _to_iso(self.schedule_last_modified_at),
424+
"ScheduleConfiguration": self.schedule_configuration,
286425
}
287426

288427
@classmethod
@@ -291,15 +430,22 @@ def from_json(cls, data: dict[str, Any]) -> "ScheduleState":
291430
# ``from_json`` hook directly. ``ScheduleConfiguration`` is an internal
292431
# type, so there is no need to route it through a (possibly custom)
293432
# converter -- keeping this hook converter-free means it round-trips
294-
# under any code path, not only the worker's threaded converter.
433+
# under any code path, not only the worker's threaded converter. Reads
434+
# accept both the .NET-compatible and legacy snake_case shapes.
295435
state = cls()
296-
state.status = ScheduleStatus(data["status"])
297-
state.execution_token = data["execution_token"]
298-
state.last_run_at = _from_iso(data.get("last_run_at"))
299-
state.next_run_at = _from_iso(data.get("next_run_at"))
300-
state.schedule_created_at = _from_iso(data.get("schedule_created_at"))
301-
state.schedule_last_modified_at = _from_iso(data.get("schedule_last_modified_at"))
302-
config_data = data.get("schedule_configuration")
436+
state.status = ScheduleStatus.from_dotnet(_get(data, "Status", "status"))
437+
# Preserve the token generated by ``__init__`` when the field is absent;
438+
# overwriting it with ``None`` would make every ``run_schedule`` signal
439+
# look stale and silently stop the schedule.
440+
token = _get(data, "ExecutionToken", "execution_token")
441+
if token is not None:
442+
state.execution_token = token
443+
state.last_run_at = _from_iso(_get(data, "LastRunAt", "last_run_at"))
444+
state.next_run_at = _from_iso(_get(data, "NextRunAt", "next_run_at"))
445+
state.schedule_created_at = _from_iso(_get(data, "ScheduleCreatedAt", "schedule_created_at"))
446+
state.schedule_last_modified_at = _from_iso(
447+
_get(data, "ScheduleLastModifiedAt", "schedule_last_modified_at"))
448+
config_data = _get(data, "ScheduleConfiguration", "schedule_configuration")
303449
state.schedule_configuration = (
304450
ScheduleConfiguration.from_json(config_data) if config_data is not None else None)
305451
return state

durabletask/scheduled/schedule_status.py

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,3 +15,52 @@ class ScheduleStatus(str, Enum):
1515

1616
PAUSED = "Paused"
1717
"""Schedule is paused."""
18+
19+
def to_dotnet_ordinal(self) -> int:
20+
"""Return the numeric value used by the .NET ``ScheduleStatus`` enum.
21+
22+
The Durable Task Scheduler dashboard reads the persisted entity state
23+
with ``System.Text.Json`` (Web defaults, no string-enum converter), so
24+
the status must be serialized as the enum's ordinal rather than its
25+
name. The ordinals match the .NET SDK order (``Uninitialized`` = 0,
26+
``Active`` = 1, ``Paused`` = 2).
27+
"""
28+
return _STATUS_TO_ORDINAL[self]
29+
30+
@classmethod
31+
def from_dotnet(cls, value: "int | str | None") -> "ScheduleStatus":
32+
"""Reconstruct a status from a persisted value.
33+
34+
Accepts the numeric ordinal written by the .NET-compatible serializer
35+
as well as the legacy string name (e.g. ``"Active"``) so that states
36+
persisted by older Python workers still round-trip.
37+
"""
38+
if isinstance(value, bool):
39+
# ``bool`` is a subclass of ``int``; reject it explicitly so a
40+
# stray boolean cannot be misread as an ordinal.
41+
return cls.UNINITIALIZED
42+
if isinstance(value, int):
43+
return _ORDINAL_TO_STATUS.get(value, cls.UNINITIALIZED)
44+
if isinstance(value, str):
45+
text = value.strip()
46+
if text.isdigit():
47+
return _ORDINAL_TO_STATUS.get(int(text), cls.UNINITIALIZED)
48+
for member in cls:
49+
if member.value.lower() == text.lower():
50+
return member
51+
# The .NET Scheduler client names the zero value "Unknown"; treat
52+
# it as the equivalent uninitialized state.
53+
if text.lower() == "unknown":
54+
return cls.UNINITIALIZED
55+
return cls.UNINITIALIZED
56+
57+
58+
_STATUS_TO_ORDINAL: dict["ScheduleStatus", int] = {
59+
ScheduleStatus.UNINITIALIZED: 0,
60+
ScheduleStatus.ACTIVE: 1,
61+
ScheduleStatus.PAUSED: 2,
62+
}
63+
64+
_ORDINAL_TO_STATUS: dict[int, "ScheduleStatus"] = {
65+
ordinal: status for status, ordinal in _STATUS_TO_ORDINAL.items()
66+
}

0 commit comments

Comments
 (0)