Use to_json/from_json hooks for schedule options

With the serializer now honoring the to_json hook for dataclasses, the schedule
option types expose to_json/from_json instead of bespoke to_dict/from_dict. The
schedule entity operations can again annotate their input as the option
dataclass and let the worker reconstruct it via the from_json hook, removing the
manual _coerce_options shim and the Any-typed parameter workaround.

Author: andystaples

durabletask/scheduled/client.py

@@ -71,11 +71,11 @@ def _run_operation(self, operation_name: str, input: Any | None = None) -> None:
 
     def create(self, options: ScheduleCreationOptions) -> None:
         """Create or update this schedule with the given configuration."""
-        self._run_operation(transitions.CREATE_SCHEDULE, options.to_dict())
+        self._run_operation(transitions.CREATE_SCHEDULE, options.to_json())
 
     def update(self, options: ScheduleUpdateOptions) -> None:
         """Update this schedule's configuration."""
-        self._run_operation(transitions.UPDATE_SCHEDULE, options.to_dict())
+        self._run_operation(transitions.UPDATE_SCHEDULE, options.to_json())
 
     def pause(self) -> None:
         """Pause this schedule."""

durabletask/scheduled/models.py

@@ -55,7 +55,7 @@ def __post_init__(self):
             raise ValueError("orchestration_name cannot be empty.")
         _validate_interval(self.interval)
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_json(self) -> dict[str, Any]:
         return {
             "schedule_id": self.schedule_id,
             "orchestration_name": self.orchestration_name,
@@ -67,9 +67,9 @@ def to_dict(self) -> dict[str, Any]:
             "start_immediately_if_late": self.start_immediately_if_late,
         }
 
-    @staticmethod
-    def from_dict(data: dict[str, Any]) -> "ScheduleCreationOptions":
-        return ScheduleCreationOptions(
+    @classmethod
+    def from_json(cls, data: dict[str, Any]) -> "ScheduleCreationOptions":
+        return cls(
             schedule_id=data["schedule_id"],
             orchestration_name=data["orchestration_name"],
             interval=timedelta(seconds=data["interval_seconds"]),
@@ -97,7 +97,7 @@ def __post_init__(self):
         if self.interval is not None:
             _validate_interval(self.interval)
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_json(self) -> dict[str, Any]:
         return {
             "orchestration_name": self.orchestration_name,
             "orchestration_input": self.orchestration_input,
@@ -108,9 +108,9 @@ def to_dict(self) -> dict[str, Any]:
             "start_immediately_if_late": self.start_immediately_if_late,
         }
 
-    @staticmethod
-    def from_dict(data: dict[str, Any]) -> "ScheduleUpdateOptions":
-        return ScheduleUpdateOptions(
+    @classmethod
+    def from_json(cls, data: dict[str, Any]) -> "ScheduleUpdateOptions":
+        return cls(
             orchestration_name=data.get("orchestration_name"),
             orchestration_input=data.get("orchestration_input"),
             orchestration_instance_id=data.get("orchestration_instance_id"),

durabletask/scheduled/schedule_entity.py

@@ -35,17 +35,6 @@ def _ensure_aware(value: datetime | None) -> datetime | None:
     return value
 
 
-def _coerce_options(input: Any, cls: type) -> Any:
-    """Coerce a round-tripped input (dict/SimpleNamespace) into the given options dataclass."""
-    if input is None or isinstance(input, cls):
-        return input
-    if isinstance(input, SimpleNamespace):
-        input = vars(input)
-    if isinstance(input, dict):
-        return cls.from_dict(input)
-    return input
-
-
 class Schedule(DurableEntity):
     """Entity that manages the state and execution of a scheduled task.
 
@@ -74,15 +63,8 @@ def _can_transition_to(self, state: ScheduleState, operation_name: str,
                            target_status: ScheduleStatus) -> bool:
         return transitions.is_valid_transition(operation_name, state.status, target_status)
 
-    # NOTE: the input is intentionally annotated ``Any`` rather than
-    # ``ScheduleCreationOptions``. The worker reconstructs an entity operation's
-    # input from its parameter annotation; a dataclass annotation would map the
-    # wire dict by field name and drop our JSON-friendly fields (e.g.
-    # ``interval_seconds``). Keeping ``Any`` lets the raw dict reach
-    # ``_coerce_options``, which rebuilds the options via ``from_dict``.
-    def create_schedule(self, options: Any) -> None:
+    def create_schedule(self, options: ScheduleCreationOptions) -> None:
         """Create a new schedule. If one already exists, update it in place."""
-        options = _coerce_options(options, ScheduleCreationOptions)
         state = self._load_state()
 
         if not self._can_transition_to(state, transitions.CREATE_SCHEDULE, ScheduleStatus.ACTIVE):
@@ -111,10 +93,8 @@ def create_schedule(self, options: Any) -> None:
             state.execution_token,
         )
 
-    # NOTE: input annotated ``Any`` for the same reason as ``create_schedule``.
-    def update_schedule(self, options: Any) -> None:
+    def update_schedule(self, options: ScheduleUpdateOptions) -> None:
         """Update an existing schedule's configuration."""
-        options = _coerce_options(options, ScheduleUpdateOptions)
         state = self._load_state()
 
         if not self._can_transition_to(state, transitions.UPDATE_SCHEDULE, state.status):

tests/durabletask/scheduled/test_models.py

@@ -51,8 +51,8 @@ def test_round_trip_through_json(self):
             orchestration_input={"key": "value"}, orchestration_instance_id="inst-1",
             start_at=start, end_at=end, start_immediately_if_late=True)
 
-        encoded = shared.to_json(options.to_dict())
-        decoded = ScheduleCreationOptions.from_dict(shared.from_json(encoded))
+        encoded = shared.to_json(options)
+        decoded = shared.from_json(encoded, ScheduleCreationOptions)
 
         assert decoded.schedule_id == "s1"
         assert decoded.orchestration_name == "orch"
@@ -71,7 +71,7 @@ def test_interval_validation(self):
 
     def test_round_trip_through_json(self):
         options = ScheduleUpdateOptions(orchestration_name="orch2", interval=timedelta(seconds=10))
-        decoded = ScheduleUpdateOptions.from_dict(shared.from_json(shared.to_json(options.to_dict())))
+        decoded = shared.from_json(shared.to_json(options), ScheduleUpdateOptions)
         assert decoded.orchestration_name == "orch2"
         assert decoded.interval == timedelta(seconds=10)
         assert decoded.start_at is None

tests/durabletask/scheduled/test_schedule_entity.py

@@ -59,7 +59,7 @@ def _start_actions(actions):
 def _creation_options(**kwargs):
     base = dict(schedule_id=SCHEDULE_ID, orchestration_name="my_orch", interval=timedelta(seconds=30))
     base.update(kwargs)
-    return ScheduleCreationOptions(**base).to_dict()
+    return ScheduleCreationOptions(**base)
 
 
 class TestCreate:
@@ -110,15 +110,15 @@ def test_update_changes_config_and_resignals(self):
         h = Harness()
         h.run("create_schedule", _creation_options())
         _, actions = h.run("update_schedule",
-                           ScheduleUpdateOptions(interval=timedelta(seconds=120)).to_dict())
+                           ScheduleUpdateOptions(interval=timedelta(seconds=120)))
         assert abs(h.state_dict["schedule_configuration"]["interval_seconds"] - 120) < 0.001
         assert len(_signal_actions(actions)) == 1
 
     def test_update_no_change_does_not_signal(self):
         h = Harness()
         h.run("create_schedule", _creation_options())
         _, actions = h.run("update_schedule",
-                           ScheduleUpdateOptions(orchestration_name="my_orch").to_dict())
+                           ScheduleUpdateOptions(orchestration_name="my_orch"))
         assert len(_signal_actions(actions)) == 0