Update deployment SDK to use slugs

src/prefect/cli/deploy.py

@@ -942,6 +942,7 @@ def _schedule_config_to_deployment_schedule(
     timezone = schedule_config.get("timezone")
     schedule_active = schedule_config.get("active", True)
     parameters = schedule_config.get("parameters", {})
+    slug = schedule_config.get("slug")
 
     if cron := schedule_config.get("cron"):
         cron_kwargs = {"cron": cron, "timezone": timezone}
@@ -974,6 +975,7 @@ def _schedule_config_to_deployment_schedule(
         schedule=schedule,
         active=schedule_active,
         parameters=parameters,
+        slug=slug,
     )
 
 

src/prefect/client/schemas/actions.py

@@ -103,6 +103,10 @@ class DeploymentScheduleCreate(ActionBaseModel):
         default_factory=dict,
         description="Parameter overrides for the schedule.",
     )
+    slug: Optional[str] = Field(
+        default=None,
+        description="A unique identifier for the schedule.",
+    )
 
     @field_validator("active", mode="wrap")
     @classmethod
@@ -132,6 +136,7 @@ def from_schedule(cls, schedule: Schedule) -> "DeploymentScheduleCreate":
                 ),
                 parameters=schedule.parameters,
                 active=schedule.active,
+                slug=schedule.slug,
             )
         elif schedule.cron is not None:
             return cls(
@@ -142,6 +147,7 @@ def from_schedule(cls, schedule: Schedule) -> "DeploymentScheduleCreate":
                 ),
                 parameters=schedule.parameters,
                 active=schedule.active,
+                slug=schedule.slug,
             )
         elif schedule.rrule is not None:
             return cls(
@@ -151,6 +157,7 @@ def from_schedule(cls, schedule: Schedule) -> "DeploymentScheduleCreate":
                 ),
                 parameters=schedule.parameters,
                 active=schedule.active,
+                slug=schedule.slug,
             )
         else:
             return cls(
@@ -174,6 +181,10 @@ class DeploymentScheduleUpdate(ActionBaseModel):
         default=None,
         description="Parameter overrides for the schedule.",
     )
+    slug: Optional[str] = Field(
+        default=None,
+        description="A unique identifier for the schedule.",
+    )
 
     @field_validator("max_scheduled_runs")
     @classmethod

src/prefect/client/schemas/objects.py

@@ -1098,6 +1098,10 @@ class DeploymentSchedule(ObjectBaseModel):
         default_factory=dict,
         description="Parameter overrides for the schedule.",
     )
+    slug: Optional[str] = Field(
+        default=None,
+        description="A unique identifier for the schedule.",
+    )
 
 
 class Deployment(ObjectBaseModel):

src/prefect/schedules.py

@@ -37,6 +37,7 @@ class Schedule:
             the weekday.
         active: Whether or not the schedule is active.
         parameters: A dictionary containing parameter overrides for the schedule.
+        slug: A unique identifier for the schedule.
     """
 
     interval: datetime.timedelta | None = None
@@ -49,6 +50,7 @@ class Schedule:
     day_or: bool = False
     active: bool = True
     parameters: dict[str, Any] = dataclasses.field(default_factory=dict)
+    slug: str | None = None
 
     def __post_init__(self) -> None:
         defined_fields = [
@@ -74,6 +76,7 @@ def Cron(
     day_or: bool = False,
     active: bool = True,
     parameters: dict[str, Any] | None = None,
+    slug: str | None = None,
 ) -> Schedule:
     """
     Creates a cron schedule.
@@ -89,6 +92,7 @@ def Cron(
             the weekday.
         active: Whether or not the schedule is active.
         parameters: A dictionary containing parameter overrides for the schedule.
+        slug: A unique identifier for the schedule.
 
     Returns:
         A cron schedule.
@@ -117,6 +121,7 @@ def Cron(
         day_or=day_or,
         active=active,
         parameters=parameters,
+        slug=slug,
     )
 
 
@@ -127,6 +132,7 @@ def Interval(
     timezone: str | None = None,
     active: bool = True,
     parameters: dict[str, Any] | None = None,
+    slug: str | None = None,
 ) -> Schedule:
     """
     Creates an interval schedule.
@@ -138,6 +144,7 @@ def Interval(
         timezone: A valid timezone string in IANA tzdata format (e.g. America/New_York).
         active: Whether or not the schedule is active.
         parameters: A dictionary containing parameter overrides for the schedule.
+        slug: A unique identifier for the schedule.
 
     Returns:
         An interval schedule.
@@ -173,6 +180,7 @@ def Interval(
         timezone=timezone,
         active=active,
         parameters=parameters,
+        slug=slug,
     )
 
 
@@ -182,6 +190,7 @@ def RRule(
     timezone: str | None = None,
     active: bool = True,
     parameters: dict[str, Any] | None = None,
+    slug: str | None = None,
 ) -> Schedule:
     """
     Creates an RRule schedule.
@@ -191,6 +200,7 @@ def RRule(
         timezone: A valid timezone string in IANA tzdata format (e.g. America/New_York).
         active: Whether or not the schedule is active.
         parameters: A dictionary containing parameter overrides for the schedule.
+        slug: A unique identifier for the schedule.
 
     Returns:
         An RRule schedule.
@@ -217,4 +227,5 @@ def RRule(
         timezone=timezone,
         active=active,
         parameters=parameters,
+        slug=slug,
     )

tests/cli/test_deploy.py

@@ -2368,6 +2368,7 @@ async def test_deployment_yaml_cron_schedule(
         deploy_config["deployments"][0]["schedule"]["parameters"] = {
             "number": 42,
         }
+        deploy_config["deployments"][0]["schedule"]["slug"] = "test-slug"
 
         with prefect_file.open(mode="w") as f:
             yaml.safe_dump(deploy_config, f)
@@ -2387,6 +2388,7 @@ async def test_deployment_yaml_cron_schedule(
         assert schedule.cron == "0 4 * * *"
         assert schedule.timezone == "America/Chicago"
         assert deployment.schedules[0].parameters == {"number": 42}
+        assert deployment.schedules[0].slug == "test-slug"
 
     @pytest.mark.usefixtures("project_dir")
     async def test_deployment_yaml_cron_schedule_timezone_cli(
@@ -2456,6 +2458,7 @@ async def test_interval_schedule_deployment_yaml(self, prefect_client, work_pool
         deploy_config["deployments"][0]["schedule"]["parameters"] = {
             "number": 42,
         }
+        deploy_config["deployments"][0]["schedule"]["slug"] = "test-slug"
 
         with prefect_yaml.open(mode="w") as f:
             yaml.safe_dump(deploy_config, f)
@@ -2477,6 +2480,7 @@ async def test_interval_schedule_deployment_yaml(self, prefect_client, work_pool
         assert schedule.anchor_date == pendulum.parse("2040-02-02")
         assert schedule.timezone == "America/Chicago"
         assert deployment.schedules[0].parameters == {"number": 42}
+        assert deployment.schedules[0].slug == "test-slug"
 
     @pytest.mark.usefixtures("project_dir")
     async def test_parsing_rrule_schedule_string_literal(
@@ -2515,6 +2519,7 @@ async def test_rrule_deployment_yaml(
         deploy_config["deployments"][0]["schedule"]["parameters"] = {
             "number": 42,
         }
+        deploy_config["deployments"][0]["schedule"]["slug"] = "test-slug"
 
         with prefect_file.open(mode="w") as f:
             yaml.safe_dump(deploy_config, f)
@@ -2536,6 +2541,7 @@ async def test_rrule_deployment_yaml(
             == "DTSTART:20220910T110000\nRRULE:FREQ=HOURLY;BYDAY=MO,TU,WE,TH,FR,SA;BYHOUR=9,10,11,12,13,14,15,16,17"
         )
         assert deployment.schedules[0].parameters == {"number": 42}
+        assert deployment.schedules[0].slug == "test-slug"
 
     @pytest.mark.usefixtures("project_dir")
     async def test_can_provide_multiple_schedules_via_command(

tests/runner/test_runner.py

@@ -1930,6 +1930,7 @@ async def test_deploy(
                 schedule=Interval(
                     3600,
                     parameters={"number": 42},
+                    slug="test-slug",
                 ),
             ),
             await (
@@ -1941,6 +1942,7 @@ async def test_deploy(
                 schedule=Interval(
                     3600,
                     parameters={"number": 42},
+                    slug="test-slug",
                 ),
             ),
             work_pool_name=work_pool_with_image_variable.name,
@@ -1971,6 +1973,7 @@ async def test_deploy(
             seconds=3600
         )
         assert deployment_1.schedules[0].parameters == {"number": 42}
+        assert deployment_1.schedules[0].slug == "test-slug"
 
         deployment_2 = await prefect_client.read_deployment_by_name(
             "test-flow/test_runner"
@@ -1983,6 +1986,7 @@ async def test_deploy(
             seconds=3600
         )
         assert deployment_2.schedules[0].parameters == {"number": 42}
+        assert deployment_2.schedules[0].slug == "test-slug"
 
         console_output = capsys.readouterr().out
         assert "prefect worker start --pool" in console_output

tests/test_flows.py

@@ -4370,9 +4370,14 @@ def add_two(number: int):
                 Interval(
                     3600,
                     parameters={"number": 42},
+                    slug="test-interval-schedule",
+                ),
+                Cron("* * * * *", parameters={"number": 42}, slug="test-cron-schedule"),
+                RRule(
+                    "FREQ=MINUTELY",
+                    parameters={"number": 42},
+                    slug="test-rrule-schedule",
                 ),
-                Cron("* * * * *", parameters={"number": 42}),
-                RRule("FREQ=MINUTELY", parameters={"number": 42}),
             ],
         )
 
@@ -4385,6 +4390,14 @@ def add_two(number: int):
 
         assert all(parameters == {"number": 42} for parameters in all_parameters)
 
+        expected_slugs = {
+            "test-interval-schedule",
+            "test-cron-schedule",
+            "test-rrule-schedule",
+        }
+        actual_slugs = {schedule.slug for schedule in deployment.schedules}
+        assert actual_slugs == expected_slugs
+
     @pytest.mark.parametrize(
         "kwargs",
         [
@@ -4394,6 +4407,14 @@ def add_two(number: int):
                     {"interval": 3600},
                     {"cron": "* * * * *"},
                     {"rrule": "FREQ=MINUTELY"},
+                    {
+                        "schedules": [
+                            Interval(3600, slug="test-interval-schedule"),
+                            Cron("* * * * *", slug="test-cron-schedule"),
+                            RRule("FREQ=MINUTELY", slug="test-rrule-schedule"),
+                        ]
+                    },
+                    {"schedule": Interval(3600, slug="test-interval-schedule")},
                 ],
                 2,
             )