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

Allow users to specify timezone when setting up CRON triggers to enable
scheduling in local time zones with automatic DST handling.

Changes:
- Add timezone field to CronTrigger and DatabaseTriggers models
- Implement timezone-aware cron scheduling using Python's zoneinfo
- Validate timezone using IANA timezone database
- Update Python SDK to support timezone parameter
- Add comprehensive documentation with examples
- Default to UTC for backward compatibility

The timezone field is optional and defaults to "UTC". All trigger times
are internally stored in UTC while croniter calculations respect the
specified timezone, ensuring correct scheduling across time zones and
DST transitions.

Signed-off-by: Sparsh <sparsh.raj30@gmail.com>

Author: spa-raj

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 ZoneInfo, 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()
-    
+
     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,