Repository: PrefectHQ/prefect
Pull request: Fix support for legacy schedule in `build_from_flow`
The recent changes for enhanced scheduling broke the way the legacy `schedule` parameter worked for `build_from_flow`. This ensures that the legacy field is able to add and remove schedules, but in the case that both `schedule` and `schedules` are provided `schedules` will be the source of truth.

Closes #12255 

### Checklist
<!-- These boxes may be checked after opening the pull request. -->

- [x] This pull request references any related issue by including "closes `<link to issue>`"
	- If no issue exists and your change is not a small fix, please [create an issue](https://github.com/PrefectHQ/prefect/issues/new/choose) first.
- [x] This pull request includes tests or only affects documentation.
- [x] This pull request includes a label categorizing the change e.g. `maintenance`, `fix`, `feature`, `enhancement`, `docs`.
  <!-- If you do not have permission to add a label, a maintainer will add one for you -->

For documentation changes:

- [ ] This pull request includes redirect settings in `netlify.toml` for files that are removed or renamed.

For new functions or classes in the Python SDK:

- [ ] This pull request includes helpful docstrings.
- [ ] If a new Python file was added, this pull request contains a stub page in the Python SDK docs and an entry in `mkdocs.yml` navigation.

Problem statement:
"schedule" not recognised in Deployment.build_from_flow
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar issue and didn't find it.
- [X] I searched the Prefect documentation for this issue.
- [X] I checked that this issue is related to Prefect and not one of its dependencies.

### Bug summary

When setting a schedule using build_from_flow, the schedule is ignored. 

### Reproduction

```python3
from prefect import flow
from prefect.deployments import Deployment

from prefect.client.schemas.schedules import CronSchedule


@flow(log_prints=True)
def local_flow():
    print("I'm a flow written to test deploy from source")


if __name__ == "__main__":
    deployment = Deployment.build_from_flow(
        name='test-sched',
        flow=local_flow,
        work_queue_name="managed",
        schedule=CronSchedule(cron="0 0 * * *", timezone="America/Chicago"),
    )
    deployment.apply()
```


### Error

```python3
No error.  The schedule deprecation message is still shown.  No schedule shows in the UI.
```


### Versions

```Text
Version:             2.16.3
API version:         0.8.4
Python version:      3.10.4
Git commit:          e3f02c00
Built:               Thu, Mar 7, 2024 4:56 PM
OS/Arch:             darwin/x86_64
Profile:             staging
Server type:         cloud
```


### Additional context

Works as expected on 2.14.21

Hints from issue discussion:

Commit message: Fix support for legacy schedule in `build_from_flow`

Patch to review:
diff --git a/src/prefect/deployments/deployments.py b/src/prefect/deployments/deployments.py
index 80718aa83b73..a90696605ae0 100644
--- a/src/prefect/deployments/deployments.py
+++ b/src/prefect/deployments/deployments.py
@@ -38,6 +38,7 @@
 from prefect.context import FlowRunContext, PrefectObjectRegistry, TaskRunContext
 from prefect.deployments.schedules import (
     FlexibleScheduleList,
+    create_minimal_deployment_schedule,
     normalize_to_minimal_deployment_schedules,
 )
 from prefect.deployments.steps.core import run_steps
@@ -53,6 +54,7 @@
 from prefect.logging.loggers import flow_run_logger, get_logger
 from prefect.states import Scheduled
 from prefect.tasks import Task
+from prefect.utilities.annotations import NotSet
 from prefect.utilities.asyncutils import run_sync_in_worker_thread, sync_compatible
 from prefect.utilities.callables import ParameterSchema, parameter_schema
 from prefect.utilities.filesystem import relative_path_to_current_platform, tmpchdir
@@ -649,21 +651,27 @@ def validate_deprecated_schedule_fields(cls, values):
             )
         return values
 
-    @validator("schedule")
-    def validate_schedule(cls, value):
-        if value:
-            cls._validate_schedule(value)
-        return value
-
     @root_validator(pre=True)
-    def validate_schedules(cls, values):
-        if "schedules" in values:
-            values["schedules"] = normalize_to_minimal_deployment_schedules(
-                values["schedules"]
-            )
+    def reconcile_schedules(cls, values):
+        schedule = values.get("schedule", NotSet)
+        schedules = values.get("schedules", NotSet)
+
+        if schedules is not NotSet:
+            values["schedules"] = normalize_to_minimal_deployment_schedules(schedules)
+        elif schedule is not NotSet:
+            values["schedule"] = None
+
+            if schedule is None:
+                values["schedules"] = []
+            else:
+                values["schedules"] = [
+                    create_minimal_deployment_schedule(
+                        schedule=schedule, active=values.get("is_schedule_active")
+                    )
+                ]
 
-            for schedule in values["schedules"]:
-                cls._validate_schedule(schedule.schedule)
+        for schedule in values.get("schedules", []):
+            cls._validate_schedule(schedule.schedule)
 
         return values
 
@@ -984,11 +992,17 @@ async def build_from_flow(
 
         # note that `deployment.load` only updates settings that were *not*
         # provided at initialization
-        deployment = cls(
-            name=name,
-            schedules=schedules,
+
+        deployment_args = {
+            "name": name,
+            "flow_name": flow.name,
             **kwargs,
-        )
+        }
+
+        if schedules is not None:
+            deployment_args["schedules"] = schedules
+
+        deployment = cls(**deployment_args)
         deployment.flow_name = flow.name
         if not deployment.entrypoint:
             ## first see if an entrypoint can be determined
diff --git a/src/prefect/deployments/schedules.py b/src/prefect/deployments/schedules.py
index cf33e9a98926..cc1868605353 100644
--- a/src/prefect/deployments/schedules.py
+++ b/src/prefect/deployments/schedules.py
@@ -2,6 +2,7 @@
 
 from prefect.client.schemas.objects import MinimalDeploymentSchedule
 from prefect.client.schemas.schedules import SCHEDULE_TYPES
+from prefect.server.schemas.schedules import SCHEDULE_TYPES as SERVER_SCHEDULE_TYPES
 
 FlexibleScheduleList = Sequence[Union[MinimalDeploymentSchedule, dict, SCHEDULE_TYPES]]
 
@@ -28,6 +29,11 @@ def normalize_to_minimal_deployment_schedules(
                 normalized.append(create_minimal_deployment_schedule(**obj))
             elif isinstance(obj, MinimalDeploymentSchedule):
                 normalized.append(obj)
+            elif isinstance(obj, get_args(SERVER_SCHEDULE_TYPES)):
+                raise ValueError(
+                    "Server schema schedules are not supported. Please use "
+                    "the schedule objects from `prefect.client.schemas.schedules`"
+                )
             else:
                 raise ValueError(
                     "Invalid schedule provided. Must be a schedule object, a dict,"
diff --git a/tests/deployment/test_schedules.py b/tests/deployment/test_schedules.py
index a5584b4e34c5..7a7138866297 100644
--- a/tests/deployment/test_schedules.py
+++ b/tests/deployment/test_schedules.py
@@ -13,6 +13,7 @@
     create_minimal_deployment_schedule,
     normalize_to_minimal_deployment_schedules,
 )
+from prefect.server.schemas.schedules import CronSchedule as ServerCronSchedule
 
 
 @pytest.mark.parametrize(
@@ -108,6 +109,13 @@ def test_normalize_mixed():
     assert normalized[2].schedule.rrule == "FREQ=DAILY"
 
 
+def test_normalize_server_schema():
+    with pytest.raises(ValueError, match="Server schema schedules are not supported"):
+        normalize_to_minimal_deployment_schedules(
+            schedules=[ServerCronSchedule(cron="0 0 * * *")]
+        )
+
+
 def test_normalize_incompatible():
     with pytest.raises(ValueError, match="Invalid schedule provided"):
         normalize_to_minimal_deployment_schedules(schedules=[1, 2, 3])
diff --git a/tests/test_deployments.py b/tests/test_deployments.py
index bad0c67c944c..f43b37bde572 100644
--- a/tests/test_deployments.py
+++ b/tests/test_deployments.py
@@ -546,6 +546,46 @@ async def test_build_from_flow_set_schedules_shorthand(self, flow_function):
             isinstance(s, MinimalDeploymentSchedule) for s in deployment.schedules
         )
 
+    async def test_build_from_flow_legacy_schedule_supported(
+        self, flow_function, prefect_client
+    ):
+        deployment = await Deployment.build_from_flow(
+            name="legacy_schedule_supported",
+            flow=flow_function,
+            schedule=CronSchedule(cron="2 1 * * *", timezone="America/Chicago"),
+        )
+
+        deployment_id = await deployment.apply()
+
+        refreshed = await prefect_client.read_deployment(deployment_id)
+        assert refreshed.schedule.cron == "2 1 * * *"
+
+    async def test_build_from_flow_clear_schedules_via_legacy_schedule(
+        self, flow_function, prefect_client
+    ):
+        deployment = await Deployment.build_from_flow(
+            name="clear_schedules_via_legacy_schedule",
+            flow=flow_function,
+            schedule=CronSchedule(cron="2 1 * * *", timezone="America/Chicago"),
+        )
+
+        deployment_id = await deployment.apply()
+
+        refreshed = await prefect_client.read_deployment(deployment_id)
+        assert refreshed.schedule.cron == "2 1 * * *"
+
+        deployment = await Deployment.build_from_flow(
+            name="clear_schedules_via_legacy_schedule",
+            flow=flow_function,
+            schedule=None,
+        )
+
+        deployment_id_2 = await deployment.apply()
+        assert deployment_id == deployment_id_2
+
+        refreshed = await prefect_client.read_deployment(deployment_id)
+        assert refreshed.schedule is None
+
 
 class TestYAML:
     def test_deployment_yaml_roundtrip(self, tmp_path):


Code review comments:
diff_hunk: @@ -28,6 +29,11 @@ def normalize_to_minimal_deployment_schedules(
                 normalized.append(create_minimal_deployment_schedule(**obj))
             elif isinstance(obj, MinimalDeploymentSchedule):
                 normalized.append(obj)
+            elif isinstance(obj, get_args(SERVER_SCHEDULE_TYPES)):
path: src/prefect/deployments/schedules.py
text: @author:
Guarding against using the server schemas instead of the client schemas as expected.