fix fractional value handling in retry delay policy (#18200)

Author: zzstoatzz

src/prefect/_internal/schemas/validators.py

@@ -402,6 +402,18 @@ def validate_compressionlib(value: str) -> str:
 
 
 # TODO: if we use this elsewhere we can change the error message to be more generic
+@overload
+def list_length_50_or_less(v: int) -> int: ...
+
+
+@overload
+def list_length_50_or_less(v: float) -> float: ...
+
+
+@overload
+def list_length_50_or_less(v: list[int]) -> list[int]: ...
+
+
 @overload
 def list_length_50_or_less(v: list[float]) -> list[float]: ...
 
@@ -410,7 +422,9 @@ def list_length_50_or_less(v: list[float]) -> list[float]: ...
 def list_length_50_or_less(v: None) -> None: ...
 
 
-def list_length_50_or_less(v: Optional[list[float]]) -> Optional[list[float]]:
+def list_length_50_or_less(
+    v: Optional[int | float | list[int] | list[float]],
+) -> Optional[int | float | list[int] | list[float]]:
     if isinstance(v, list) and (len(v) > 50):
         raise ValueError("Can not configure more than 50 retry delays per task.")
     return v

src/prefect/client/schemas/objects.py

@@ -698,7 +698,7 @@ class TaskRunPolicy(PrefectBaseModel):
         deprecated=True,
     )
     retries: Optional[int] = Field(default=None, description="The number of retries.")
-    retry_delay: Union[None, int, list[int]] = Field(
+    retry_delay: Union[None, int, float, list[int], list[float]] = Field(
         default=None,
         description="A delay time or list of delay times between retries, in seconds.",
     )
@@ -728,8 +728,8 @@ def populate_deprecated_fields(self):
     @field_validator("retry_delay")
     @classmethod
     def validate_configured_retry_delays(
-        cls, v: Optional[list[float]]
-    ) -> Optional[list[float]]:
+        cls, v: Optional[int | float | list[int] | list[float]]
+    ) -> Optional[int | float | list[int] | list[float]]:
         return list_length_50_or_less(v)
 
     @field_validator("retry_jitter_factor")

src/prefect/server/schemas/core.py

@@ -34,6 +34,7 @@
 
 from prefect._internal.schemas.validators import (
     get_or_create_run_name,
+    list_length_50_or_less,
     set_run_policy_deprecated_fields,
     validate_cache_key_length,
     validate_default_queue_id_not_none,
@@ -356,7 +357,7 @@ class TaskRunPolicy(PrefectBaseModel):
         deprecated=True,
     )
     retries: Optional[int] = Field(default=None, description="The number of retries.")
-    retry_delay: Union[None, int, List[int]] = Field(
+    retry_delay: Union[None, int, float, List[int], List[float]] = Field(
         default=None,
         description="A delay time or list of delay times between retries, in seconds.",
     )
@@ -371,11 +372,9 @@ def populate_deprecated_fields(cls, values: dict[str, Any]) -> dict[str, Any]:
     @field_validator("retry_delay")
     @classmethod
     def validate_configured_retry_delays(
-        cls, v: int | list[int] | None
-    ) -> int | list[int] | None:
-        if isinstance(v, list) and (len(v) > 50):
-            raise ValueError("Can not configure more than 50 retry delays per task.")
-        return v
+        cls, v: int | float | list[int] | list[float] | None
+    ) -> int | float | list[int] | list[float] | None:
+        return list_length_50_or_less(v)
 
     @field_validator("retry_jitter_factor")
     @classmethod

tests/test_tasks.py

@@ -4329,6 +4329,21 @@ async def test_task_cannot_configure_negative_relative_jitter(self):
             async def insanity():
                 raise RuntimeError("try again!")
 
+    def test_task_accepts_fractional_retry_delay_seconds(self):
+        @task(retries=2, retry_delay_seconds=1.5)
+        def task_with_float_delay():
+            return "success"
+
+        @task(retries=3, retry_delay_seconds=[0.5, 1.1, 2.7])
+        def task_with_float_list_delay():
+            return "success"
+
+        assert task_with_float_delay.retries == 2
+        assert task_with_float_delay.retry_delay_seconds == 1.5
+
+        assert task_with_float_list_delay.retries == 3
+        assert task_with_float_list_delay.retry_delay_seconds == [0.5, 1.1, 2.7]
+
 
 async def test_task_run_name_is_set(prefect_client, events_pipeline):
     @task(task_run_name="fixed-name")