introduce timezone for CRON

examples/triggers/basic.py

@@ -19,7 +19,7 @@ def example_task(trigger_time: datetime, x: int = 1) -> str:
 
 custom_trigger = flyte.Trigger(
     "custom_cron",
-    flyte.Cron("0 0 * * *"),  # Runs every day
+    flyte.Cron("0 0 * * *", timezone="Europe/London"),  # Runs once every day at midnight London time
     inputs={"start_time": flyte.TriggerTime, "x": 1},
 )
 

src/flyte/_internal/runtime/trigger_serde.py

@@ -12,7 +12,7 @@
 def _to_schedule(m: Union[Cron, FixedRate], kickoff_arg_name: str | None = None) -> common_pb2.Schedule:
     if isinstance(m, Cron):
         return common_pb2.Schedule(
-            cron_expression=m.expression,
+            cron_expression=m.timezone_expression,
             kickoff_time_input_arg=kickoff_arg_name,
         )
     elif isinstance(m, FixedRate):

src/flyte/_trigger.py

@@ -2,10 +2,12 @@
 
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Dict, Mapping, Union
+from typing import Any, Dict, Literal, Mapping, Union
 
 import rich.repr
 
+DEFAULT_TIMEZONE = "America/Los_Angeles"
+
 
 class _trigger_time:
     """
@@ -33,9 +35,14 @@ class Cron:
     """
 
     expression: str
+    timezone: Literal["America/Los_Angeles", "Europe/London", "Australia/Sydney", "America/New_York"] = DEFAULT_TIMEZONE
+
+    @property
+    def timezone_expression(self) -> str:
+        return f"{self.timezone} {self.expression}"
 
     def __str__(self):
-        return f"Cron Trigger: {self.expression}"
+        return f"Cron Trigger: {self.timezone_expression}"
 
 
 @rich.repr.auto

tests/flyte/internal/runtime/test_trigger_serde.py

@@ -11,6 +11,7 @@
     process_default_inputs,
     to_task_trigger,
 )
+from flyte._trigger import DEFAULT_TIMEZONE
 from flyte.types import TypeEngine
 
 
@@ -22,15 +23,15 @@ def test_cron_without_kickoff_arg(self):
         cron = Cron("0 * * * *")
         schedule = _to_schedule(cron)
 
-        assert schedule.cron_expression == "0 * * * *"
+        assert schedule.cron_expression == f"{DEFAULT_TIMEZONE} 0 * * * *"
         assert schedule.kickoff_time_input_arg == ""
 
     def test_cron_with_kickoff_arg(self):
         """Test Cron schedule with kickoff argument"""
         cron = Cron("0 0 * * *")
         schedule = _to_schedule(cron, kickoff_arg_name="trigger_time")
 
-        assert schedule.cron_expression == "0 0 * * *"
+        assert schedule.cron_expression == f"{DEFAULT_TIMEZONE} 0 0 * * *"
         assert schedule.kickoff_time_input_arg == "trigger_time"
 
     def test_fixed_rate_without_start_time(self):
@@ -65,7 +66,15 @@ def test_cron_expression_variations(self):
         for expr in expressions:
             cron = Cron(expr)
             schedule = _to_schedule(cron)
-            assert schedule.cron_expression == expr
+            assert schedule.cron_expression == f"{DEFAULT_TIMEZONE} {expr}"
+
+    def test_cron_with_timezone(self):
+        """Test Cron schedule with custom timezone"""
+        cron = Cron("0 * * * *", timezone="Europe/London")
+        schedule = _to_schedule(cron)
+
+        assert schedule.cron_expression == "Europe/London 0 * * * *"
+        assert schedule.kickoff_time_input_arg == ""
 
 
 class TestProcessDefaultInputs:
@@ -199,7 +208,7 @@ async def test_basic_cron_trigger(self):
         assert result.name == "test_trigger"
         assert result.spec.active is True
         assert result.automation_spec.type == common_pb2.TriggerAutomationSpec.Type.TYPE_SCHEDULE
-        assert result.automation_spec.schedule.cron_expression == "0 * * * *"
+        assert result.automation_spec.schedule.cron_expression == f"{DEFAULT_TIMEZONE} 0 * * * *"
 
     @pytest.mark.asyncio
     async def test_fixed_rate_trigger(self):
@@ -440,7 +449,7 @@ async def test_automation_spec_cron(self):
         result = await to_task_trigger(trigger, "test_task", task_inputs, [])
 
         assert result.automation_spec.type == common_pb2.TriggerAutomationSpec.Type.TYPE_SCHEDULE
-        assert result.automation_spec.schedule.cron_expression == "*/5 * * * *"
+        assert result.automation_spec.schedule.cron_expression == f"{DEFAULT_TIMEZONE} */5 * * * *"
         assert result.automation_spec.schedule.kickoff_time_input_arg == ""
 
     @pytest.mark.asyncio
@@ -494,7 +503,7 @@ async def test_automation_spec_with_kickoff_arg(self):
 
         result = await to_task_trigger(trigger, "test_task", task_inputs, [])
 
-        assert result.automation_spec.schedule.cron_expression == "0 12 * * *"
+        assert result.automation_spec.schedule.cron_expression == f"{DEFAULT_TIMEZONE} 0 12 * * *"
         assert result.automation_spec.schedule.kickoff_time_input_arg == "scheduled_at"