feat: add timezone support for CRON triggers (#438)

docs/docs/exosphere/triggers.md

@@ -61,13 +61,15 @@ Define triggers in your graph template:
     {
       "type": "CRON",
       "value": {
-        "expression": "0 9 * * 1-5"
+        "expression": "0 9 * * 1-5",
+        "timezone": "America/New_York"
       }
     },
     {
       "type": "CRON",
       "value": {
-        "expression": "0 0 * * 0"
+        "expression": "0 0 * * 0",
+        "timezone": "UTC"
       }
     }
   ],
@@ -77,6 +79,8 @@ Define triggers in your graph template:
 }
 ```
 
+**Note:** The `timezone` field is optional and defaults to `"UTC"` if not specified. Use IANA timezone names (e.g., `"America/New_York"`, `"Europe/London"`, `"Asia/Tokyo"`).
+
 ### Python SDK Example
 
 ```python
@@ -109,8 +113,8 @@ async def create_scheduled_graph():
 
     # Define triggers for automatic execution
     triggers = [
-        CronTrigger(expression="0 2 * * *"),    # Daily at 2:00 AM
-        CronTrigger(expression="0 */4 * * *")   # Every 4 hours
+        CronTrigger(expression="0 2 * * *", timezone="America/New_York"),  # Daily at 2:00 AM EST/EDT
+        CronTrigger(expression="0 */4 * * *", timezone="UTC")              # Every 4 hours UTC
     ]
 
     # Create the graph with triggers
@@ -158,7 +162,7 @@ asyncio.run(create_scheduled_graph())
 
 1. **Avoid Peak Times**: Schedule resource-intensive workflows during off-peak hours
 2. **Stagger Executions**: If you have multiple graphs, stagger their execution times
-3. **Consider Time Zones**: Cron expressions use server time (UTC by default)
+3. **Consider Time Zones**: Specify the `timezone` parameter to ensure your cron expressions run at the correct local time. If not specified, defaults to UTC.
 4. **Resource Planning**: Ensure your infrastructure can handle scheduled workloads
 
 ### Error Handling
@@ -191,11 +195,31 @@ result = await state_manager.upsert_graph(
 )
 ```
 
+## Timezone Support
+
+Triggers now support specifying a timezone for cron expressions, allowing you to schedule jobs in your local timezone:
+
+```python
+# Schedule a report to run at 9 AM New York time (handles DST automatically)
+CronTrigger(expression="0 9 * * 1-5", timezone="America/New_York")
+
+# Schedule a job at 5 PM London time
+CronTrigger(expression="0 17 * * *", timezone="Europe/London")
+
+# Schedule using UTC (default)
+CronTrigger(expression="0 12 * * *", timezone="UTC")
+```
+
+**Important Notes:**
+- Use IANA timezone names (e.g., `"America/New_York"`, `"Europe/London"`, `"Asia/Tokyo"`)
+- Timezones automatically handle Daylight Saving Time (DST) transitions
+- If no timezone is specified, defaults to `"UTC"`
+- All trigger times are internally stored in UTC for consistency
+
 ## Limitations
 
 - **CRON Only**: Currently only cron-based scheduling is supported
 - **No Manual Override**: Scheduled executions cannot be manually cancelled once triggered
-- **Time Zone**: All cron expressions are evaluated in server time (UTC)
 - **Minimum Interval**: Avoid scheduling more frequently than every minute
 
 ## Next Steps

python-sdk/exospherehost/models.py

@@ -160,4 +160,5 @@ def validate_default_values(cls, v: dict[str, str]) -> dict[str, str]:
         return normalized_dict
 
 class CronTrigger(BaseModel):
-    expression: str = Field(..., description="Cron expression for scheduling automatic graph execution. Uses standard 5-field format: minute hour day-of-month month day-of-week. Example: '0 9 * * 1-5' for weekdays at 9 AM.")
+    expression: str = Field(..., description="Cron expression for scheduling automatic graph execution. Uses standard 5-field format: minute hour day-of-month month day-of-week. Example: '0 9 * * 1-5' for weekdays at 9 AM.")
+    timezone: str = Field(default="UTC", description="Timezone for the cron expression (e.g., 'America/New_York', 'Europe/London', 'UTC'). Defaults to 'UTC'.")

python-sdk/exospherehost/statemanager.py

@@ -173,7 +173,8 @@ async def upsert_graph(self, graph_name: str, graph_nodes: list[GraphNodeModel],
                 {
                     "type": "CRON",
                     "value": {
-                        "expression": trigger.expression
+                        "expression": trigger.expression,
+                        "timezone": trigger.timezone
                     }
                 }
                 for trigger in triggers

state-manager/app/models/db/trigger.py

@@ -9,6 +9,7 @@
 class DatabaseTriggers(Document):
     type: TriggerTypeEnum = Field(..., description="Type of the trigger")
     expression: Optional[str] = Field(default=None, description="Expression of the trigger")
+    timezone: Optional[str] = Field(default="UTC", description="Timezone for the trigger")
     graph_name: str = Field(..., description="Name of the graph")
     namespace: str = Field(..., description="Namespace of the graph")
     trigger_time: datetime = Field(..., description="Trigger time of the trigger")

state-manager/app/models/trigger_models.py

@@ -1,7 +1,8 @@
 from pydantic import BaseModel, Field, field_validator, model_validator
 from enum import Enum
 from croniter import croniter
-from typing import Self
+from typing import Self, Optional
+from zoneinfo import available_timezones
 
 class TriggerTypeEnum(str, Enum):
     CRON = "CRON"
@@ -15,6 +16,7 @@ class TriggerStatusEnum(str, Enum):
 
 class CronTrigger(BaseModel):
     expression: str = Field(..., description="Cron expression for the trigger")
+    timezone: Optional[str] = Field(default="UTC", description="Timezone for the cron expression (e.g., 'America/New_York', 'Europe/London', 'UTC')")
 
     @field_validator("expression")
     @classmethod
@@ -23,6 +25,15 @@ def validate_expression(cls, v: str) -> str:
             raise ValueError("Invalid cron expression")
         return v
 
+    @field_validator("timezone")
+    @classmethod
+    def validate_timezone(cls, v: Optional[str]) -> str:
+        if v is None:
+            return "UTC"
+        if v not in available_timezones():
+            raise ValueError(f"Invalid timezone: {v}. Must be a valid IANA timezone (e.g., 'America/New_York', 'Europe/London', 'UTC')")
+        return v
+
 class Trigger(BaseModel):
     type: TriggerTypeEnum = Field(..., description="Type of the trigger")
     value: dict = Field(default_factory=dict, description="Value of the trigger")

state-manager/app/tasks/trigger_cron.py

@@ -8,6 +8,7 @@
 from pymongo import ReturnDocument
 from pymongo.errors import DuplicateKeyError
 from app.config.settings import get_settings
+from zoneinfo import ZoneInfo
 import croniter
 import asyncio
 
@@ -42,15 +43,26 @@ async def mark_as_failed(trigger: DatabaseTriggers):
 
 async def create_next_triggers(trigger: DatabaseTriggers, cron_time: datetime):
     assert trigger.expression is not None
-    iter = croniter.croniter(trigger.expression, trigger.trigger_time)
+
+    # Use the trigger's timezone, defaulting to UTC if not specified
+    tz = ZoneInfo(trigger.timezone or "UTC")
+
+    # Convert trigger_time to the specified timezone for croniter
+    trigger_time_tz = trigger.trigger_time.replace(tzinfo=ZoneInfo("UTC")).astimezone(tz)
+    iter = croniter.croniter(trigger.expression, trigger_time_tz)
 
     while True:
-        next_trigger_time = iter.get_next(datetime)
+        # Get next trigger time in the specified timezone
+        next_trigger_time_tz = iter.get_next(datetime)
+
+        # Convert back to UTC for storage
+        next_trigger_time = next_trigger_time_tz.astimezone(ZoneInfo("UTC")).replace(tzinfo=None)
 
         try:
             await DatabaseTriggers(
                 type=TriggerTypeEnum.CRON,
                 expression=trigger.expression,
+                timezone=trigger.timezone,
                 graph_name=trigger.graph_name,
                 namespace=trigger.namespace,
                 trigger_time=next_trigger_time,

state-manager/app/tasks/verify_graph.py

@@ -3,6 +3,7 @@
 
 from datetime import datetime
 from json_schema_to_pydantic import create_model
+from zoneinfo import ZoneInfo
 
 from app.models.db.graph_template_model import GraphTemplate
 from app.models.graph_template_validation_status import GraphTemplateValidationStatus
@@ -101,20 +102,36 @@ async def verify_inputs(graph_template: GraphTemplate, registered_nodes: list[Re
     return errors
 
 async def create_crons(graph_template: GraphTemplate):
-    expressions_to_create = set([trigger.value["expression"] for trigger in graph_template.triggers if trigger.type == TriggerTypeEnum.CRON])
+    # Build a map of (expression, timezone) -> trigger for deduplication
+    triggers_to_create = {}
+    for trigger in graph_template.triggers:
+        if trigger.type == TriggerTypeEnum.CRON:
+            expression = trigger.value["expression"]
+            timezone = trigger.value.get("timezone", "UTC")
+            triggers_to_create[(expression, timezone)] = trigger
+
+    current_time = datetime.now(ZoneInfo("UTC")).replace(tzinfo=None)
 
-    current_time = datetime.now()
-
     new_db_triggers = []
-    for expression in expressions_to_create:
-        iter = croniter.croniter(expression, current_time)
+    for (expression, timezone), trigger in triggers_to_create.items():
+        # Use the trigger's timezone, defaulting to UTC
+        tz = ZoneInfo(timezone)
+
+        # Get current time in the specified timezone
+        current_time_tz = current_time.replace(tzinfo=ZoneInfo("UTC")).astimezone(tz)
+        iter = croniter.croniter(expression, current_time_tz)
+
+        # Get next trigger time in the specified timezone
+        next_trigger_time_tz = iter.get_next(datetime)
+
+        # Convert back to UTC for storage (remove timezone info for storage)
+        next_trigger_time = next_trigger_time_tz.astimezone(ZoneInfo("UTC")).replace(tzinfo=None)
 
-        next_trigger_time = iter.get_next(datetime)
-
         new_db_triggers.append(
             DatabaseTriggers(
                 type=TriggerTypeEnum.CRON,
                 expression=expression,
+                timezone=timezone,
                 graph_name=graph_template.name,
                 namespace=graph_template.namespace,
                 trigger_status=TriggerStatusEnum.PENDING,

state-manager/tests/unit/models/test_trigger_models.py

@@ -0,0 +1,118 @@
+import pytest
+from pydantic import ValidationError
+from app.models.trigger_models import CronTrigger, Trigger, TriggerTypeEnum
+
+
+class TestCronTrigger:
+    """Test cases for CronTrigger model"""
+
+    def test_valid_cron_trigger_with_timezone(self):
+        """Test creating a valid cron trigger with timezone"""
+        trigger = CronTrigger(expression="0 9 * * *", timezone="America/New_York")
+        assert trigger.expression == "0 9 * * *"
+        assert trigger.timezone == "America/New_York"
+
+    def test_valid_cron_trigger_without_timezone(self):
+        """Test creating a valid cron trigger without timezone defaults to UTC"""
+        trigger = CronTrigger(expression="0 9 * * *")
+        assert trigger.expression == "0 9 * * *"
+        assert trigger.timezone == "UTC"
+
+    def test_valid_cron_trigger_with_utc_timezone(self):
+        """Test creating a valid cron trigger with UTC timezone"""
+        trigger = CronTrigger(expression="0 9 * * *", timezone="UTC")
+        assert trigger.expression == "0 9 * * *"
+        assert trigger.timezone == "UTC"
+
+    def test_valid_cron_trigger_with_europe_london(self):
+        """Test creating a valid cron trigger with Europe/London timezone"""
+        trigger = CronTrigger(expression="0 17 * * *", timezone="Europe/London")
+        assert trigger.expression == "0 17 * * *"
+        assert trigger.timezone == "Europe/London"
+
+    def test_valid_cron_trigger_with_asia_tokyo(self):
+        """Test creating a valid cron trigger with Asia/Tokyo timezone"""
+        trigger = CronTrigger(expression="30 8 * * 1-5", timezone="Asia/Tokyo")
+        assert trigger.expression == "30 8 * * 1-5"
+        assert trigger.timezone == "Asia/Tokyo"
+
+    def test_invalid_cron_expression(self):
+        """Test creating a cron trigger with invalid expression"""
+        with pytest.raises(ValidationError) as exc_info:
+            CronTrigger(expression="invalid cron", timezone="UTC")
+
+        errors = exc_info.value.errors()
+        assert len(errors) == 1
+        assert "Invalid cron expression" in str(errors[0]["ctx"]["error"])
+
+    def test_invalid_timezone(self):
+        """Test creating a cron trigger with invalid timezone"""
+        with pytest.raises(ValidationError) as exc_info:
+            CronTrigger(expression="0 9 * * *", timezone="Invalid/Timezone")
+
+        errors = exc_info.value.errors()
+        assert len(errors) == 1
+        assert "Invalid timezone" in str(errors[0]["ctx"]["error"])
+        assert "Invalid/Timezone" in str(errors[0]["ctx"]["error"])
+
+    def test_none_timezone_defaults_to_utc(self):
+        """Test that None timezone defaults to UTC"""
+        trigger = CronTrigger(expression="0 9 * * *", timezone=None)
+        assert trigger.timezone == "UTC"
+
+    def test_complex_cron_expression_with_timezone(self):
+        """Test complex cron expression with timezone"""
+        trigger = CronTrigger(expression="0 0 1,15 * *", timezone="America/Los_Angeles")
+        assert trigger.expression == "0 0 1,15 * *"
+        assert trigger.timezone == "America/Los_Angeles"
+
+    def test_every_15_minutes_cron_with_timezone(self):
+        """Test every 15 minutes cron with timezone"""
+        trigger = CronTrigger(expression="*/15 * * * *", timezone="Europe/Paris")
+        assert trigger.expression == "*/15 * * * *"
+        assert trigger.timezone == "Europe/Paris"
+
+
+class TestTrigger:
+    """Test cases for Trigger model"""
+
+    def test_valid_trigger_with_cron_and_timezone(self):
+        """Test creating a valid trigger with CRON type and timezone"""
+        trigger = Trigger(
+            type=TriggerTypeEnum.CRON,
+            value={"expression": "0 9 * * *", "timezone": "America/New_York"}
+        )
+        assert trigger.type == TriggerTypeEnum.CRON
+        assert trigger.value["expression"] == "0 9 * * *"
+        assert trigger.value["timezone"] == "America/New_York"
+
+    def test_valid_trigger_with_cron_without_timezone(self):
+        """Test creating a valid trigger with CRON type without timezone"""
+        trigger = Trigger(
+            type=TriggerTypeEnum.CRON,
+            value={"expression": "0 9 * * *"}
+        )
+        assert trigger.type == TriggerTypeEnum.CRON
+        assert trigger.value["expression"] == "0 9 * * *"
+
+    def test_invalid_trigger_with_invalid_cron_expression(self):
+        """Test creating a trigger with invalid cron expression"""
+        with pytest.raises(ValidationError) as exc_info:
+            Trigger(
+                type=TriggerTypeEnum.CRON,
+                value={"expression": "invalid cron"}
+            )
+
+        errors = exc_info.value.errors()
+        assert len(errors) > 0
+
+    def test_invalid_trigger_with_invalid_timezone(self):
+        """Test creating a trigger with invalid timezone"""
+        with pytest.raises(ValidationError) as exc_info:
+            Trigger(
+                type=TriggerTypeEnum.CRON,
+                value={"expression": "0 9 * * *", "timezone": "Invalid/Zone"}
+            )
+
+        errors = exc_info.value.errors()
+        assert len(errors) > 0