Toggle set schedule active intelligently

changes/pr37.yaml

@@ -0,0 +1,2 @@
+enhancement:
+  - "Set schedule to inactive if neither Flow nor Flow Group have a schedule - [#37](https://github.com/PrefectHQ/server/pull/37)"

src/prefect_server/api/flows.py

@@ -184,7 +184,7 @@ async def create_flow(
         config.core_version_cutoff
     ):
         raise ValueError(
-            "Prefect Serve requires new flows to be built with Prefect "
+            "Prefect Server requires new flows to be built with Prefect "
             f"{config.core_version_cutoff}+, but this flow was built with "
             f"Prefect {core_version}."
         )
@@ -234,7 +234,7 @@ async def create_flow(
                 {"name": {"_eq": version_group_id}},
             ]
         }
-    ).first()
+    ).first({"id", "schedule"})
     if flow_group is None:
         flow_group_id = await models.FlowGroup(
             tenant_id=tenant_id,
@@ -250,6 +250,11 @@ async def create_flow(
 
     version = (await models.Flow.where(version_where).max({"version"}))["version"] or 0
 
+    # if there is no referenceable schedule for this Flow,
+    # we should set its "schedule" to inactive to avoid confusion
+    if flow.schedule is None and getattr(flow_group, "schedule", None) is None:
+        set_schedule_active = False
+
     # precompute task ids to make edges easy to add to database
     flow_id = await models.Flow(
         tenant_id=tenant_id,

tests/api/test_flows.py

@@ -53,6 +53,42 @@ async def test_create_flow(self, project_id, flow):
         )
         assert await models.Flow.exists(flow_id)
 
+    async def test_create_flow_with_no_schedule_sets_schedule_inactive(
+        self, project_id, flow
+    ):
+        assert flow.schedule is None
+
+        flow_id = await api.flows.create_flow(
+            project_id=project_id, serialized_flow=flow.serialize()
+        )
+
+        flow = await models.Flow.where(id=flow_id).first({"is_schedule_active"})
+        assert flow.is_schedule_active is False
+
+    async def test_create_flow_with_only_flow_group_schedule_keeps_schedule_active(
+        self, project_id, flow_group_id
+    ):
+        success = await api.flow_groups.set_flow_group_schedule(
+            flow_group_id=flow_group_id,
+            clocks=[{"type": "CronClock", "cron": "42 0 0 * * *"}],
+        )
+        assert success is True
+
+        flow_group = await models.FlowGroup.where(id=flow_group_id).first(
+            {"schedule", "name"}
+        )
+        assert flow_group.schedule is not None
+
+        flow = prefect.Flow("empty Flow")
+        flow_id = await api.flows.create_flow(
+            project_id=project_id,
+            serialized_flow=flow.serialize(),
+            version_group_id=flow_group.name,
+        )
+
+        flow = await models.Flow.where(id=flow_id).first({"is_schedule_active"})
+        assert flow.is_schedule_active is True
+
     async def test_create_old_and_valid_flow(self, project_id, flow):
         serialized_flow = flow.serialize()
         serialized_flow["environment"]["__version__"] = "0.0.42"