Deprecate cron_expression

flytekit/core/schedule.py

@@ -85,14 +85,11 @@ def my_wf(kickoff_time: datetime): ...
                 kickoff_time_input_arg="kickoff_time")
 
         """
-        if cron_expression is None and schedule is None:
-            raise AssertionError("Either `cron_expression` or `schedule` should be specified.")
-
-        if cron_expression is not None and offset is not None:
-            raise AssertionError("Only `schedule` is supported when specifying `offset`.")
-
-        if cron_expression is not None:
-            CronSchedule._validate_expression(cron_expression)
+        if cron_expression:
+            raise AssertionError(
+                "cron_expression is deprecated and should not be used. Use `schedule` instead. "
+                "See the documentation for more information."
+            )
 
         if schedule is not None:
             CronSchedule._validate_schedule(schedule)

tests/flytekit/unit/core/test_launch_plan.py

@@ -44,7 +44,7 @@ def wf(a: int, c: str) -> (int, str):
     # fixed_and_default_end
 
     # schedule_start
-    sched = CronSchedule("* * ? * * *", kickoff_time_input_arg="abc")
+    sched = CronSchedule(schedule="*/1 * * * *", kickoff_time_input_arg="abc")
     email_notif = notification.Email(
         phases=[_execution_model.WorkflowExecutionPhase.SUCCEEDED], recipients_email=["[email protected]"]
     )
@@ -122,7 +122,7 @@ def wf(a: int, c: str) -> (int, str):
         launch_plan.LaunchPlan.get_or_create(workflow=wf, name="get_or_create_fixed")
 
     # Schedule Parameter
-    obj = CronSchedule("* * ? * * *", kickoff_time_input_arg="abc")
+    obj = CronSchedule(schedule="*/1 * * * *", kickoff_time_input_arg="abc")
     schedule_lp = launch_plan.LaunchPlan.get_or_create(workflow=wf, name="get_or_create_schedule", schedule=obj)
     schedule_lp1 = launch_plan.LaunchPlan.get_or_create(workflow=wf, name="get_or_create_schedule", schedule=obj)
 
@@ -323,8 +323,8 @@ def wf(a: int, c: str) -> str:
         u = t2(a=x, b=y, c=c)
         return u
 
-    obj = CronSchedule("* * ? * * *", kickoff_time_input_arg="abc")
-    obj1 = CronSchedule("10 * ? * * *", kickoff_time_input_arg="abc")
+    obj = CronSchedule(schedule="* * * * * *", kickoff_time_input_arg="abc")
+    obj1 = CronSchedule(schedule="10 * * * * *", kickoff_time_input_arg="abc")
     slack_notif = notification.Slack(
         phases=[_execution_model.WorkflowExecutionPhase.SUCCEEDED], recipients_email=["[email protected]"]
     )

tests/flytekit/unit/core/test_schedule.py

@@ -10,24 +10,26 @@
 
 
 def test_cron():
-    obj = CronSchedule("* * ? * * *", kickoff_time_input_arg="abc")
-    assert obj.kickoff_time_input_arg == "abc"
-    assert obj.cron_expression == "* * ? * * *"
-    assert obj == CronSchedule.from_flyte_idl(obj.to_flyte_idl())
+    with pytest.raises(AssertionError):
+        obj = CronSchedule("* * ? * * *", kickoff_time_input_arg="abc")
+        assert obj.kickoff_time_input_arg == "abc"
+        assert obj.cron_expression == "* * ? * * *"
+        assert obj == CronSchedule.from_flyte_idl(obj.to_flyte_idl())
 
 
 def test_cron_karg():
-    obj = CronSchedule(cron_expression="* * ? * * *", kickoff_time_input_arg="abc")
-    assert obj.kickoff_time_input_arg == "abc"
-    assert obj.cron_expression == "* * ? * * *"
-    assert obj == CronSchedule.from_flyte_idl(obj.to_flyte_idl())
+    with pytest.raises(AssertionError):
+        obj = CronSchedule(cron_expression="* * ? * * *", kickoff_time_input_arg="abc")
+        assert obj.kickoff_time_input_arg == "abc"
+        assert obj.cron_expression == "* * ? * * *"
+        assert obj == CronSchedule.from_flyte_idl(obj.to_flyte_idl())
 
 
 def test_cron_validation():
-    with pytest.raises(ValueError):
+    with pytest.raises(AssertionError):
         CronSchedule("* * * * * *", kickoff_time_input_arg="abc")
 
-    with pytest.raises(ValueError):
+    with pytest.raises(AssertionError):
         CronSchedule("* * ? * *", kickoff_time_input_arg="abc")