change from regex to python for validation

tests/monitor/manager/test_monitor_setup.py

@@ -287,27 +287,41 @@ def test_cron_schedule_for_analyzer(monitor_setup) -> None:
         cron="0 0 * * *"
     )
 
-    monitor_setup.schedule = CronSchedule(cron="0 0 * * 1-5") # Monday-Friday
+    monitor_setup.schedule = CronSchedule(cron="0 0 * * 1-5")
     monitor_setup.apply()
 
     assert monitor_setup.analyzer.schedule == CronSchedule(
         cron="0 0 * * 1-5"
     )
 
-    monitor_setup.schedule = CronSchedule(cron="0 0 * * 6,0") # Weekends only
+    monitor_setup.schedule = CronSchedule(cron="0 0 * * 6,0")
     monitor_setup.apply()
 
     assert monitor_setup.analyzer.schedule == CronSchedule(
         cron="0 0 * * 6,0"
     )
 
-    monitor_setup.schedule = CronSchedule(cron="0 9-17 * * *") # Hours 9-17 (Hourly models only)
+    monitor_setup.schedule = CronSchedule(cron="0 9-17 * * *")
     monitor_setup.apply()
 
     assert monitor_setup.analyzer.schedule == CronSchedule(
         cron="0 9-17 * * *"
     )
 
+    monitor_setup.schedule = CronSchedule(cron="0 9,10,17 * * *") 
+    monitor_setup.apply()
+
+    assert monitor_setup.analyzer.schedule == CronSchedule(
+        cron="0 9,10,17 * * *"
+    )
+
+    monitor_setup.schedule = CronSchedule(cron="0,1 9,10,17 1,2,3 2,4,5 2,4")
+    monitor_setup.apply()
+
+    assert monitor_setup.analyzer.schedule == CronSchedule(
+        cron="0,1 9,10,17 1,2,3 2,4,5 2,4"
+    )
+
     # All below Must fail
 
     monitor_setup.schedule = CronSchedule(cron="* * * * *") # Every minute

whylabs_toolkit/helpers/cron_validators.py

@@ -0,0 +1,53 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class SplitCron:
+    day_of_week: str
+    month: str
+    day_of_month: str
+    hour: str
+    minute: str
+
+
+def split_cron_expression(cron: str) -> SplitCron:
+    """Split the cron expression into its components."""
+    cron_slots = cron.split(" ")
+    if len(cron_slots) != 5:
+        raise ValueError("CronSchedule must have 5 fields.")
+    return SplitCron(
+        day_of_week=cron_slots[0],
+        month=cron_slots[1],
+        day_of_month=cron_slots[2],
+        hour=cron_slots[3],
+        minute=cron_slots[4],
+    )
+
+
+def _is_not_less_granular_than_1_hour(split_cron: SplitCron) -> bool:
+    """Check if the cron expression is less granular than 1 hour."""
+    # Specific days checks
+    if split_cron.minute != "*" and split_cron.minute != "0":
+        return True
+    if split_cron.hour != "*" and split_cron.hour != "0":
+        return True
+    if split_cron.day_of_month != "*" and split_cron.day_of_month != "1":
+        return True
+    if split_cron.month != "*" and split_cron.month != "1":
+        return True
+    if split_cron.day_of_week != "*" and split_cron.day_of_week != "1":
+        return True
+
+    # Check range
+    for field in (split_cron.day_of_week, split_cron.month, split_cron.day_of_month, split_cron.hour):
+        for item in field.split(","):
+            if "-" in item:
+                start, end = map(int, item.split("-"))
+                if end - start > 0:
+                    return False
+    return False
+
+
+def validate_cron_expression(cron: str) -> bool:
+    split_cron = split_cron_expression(cron)
+    return _is_not_less_granular_than_1_hour(split_cron=split_cron)

whylabs_toolkit/monitor/models/analyzer/analyzer.py

@@ -1,5 +1,4 @@
 """Schema for analyses."""
-import re
 from typing import Any, Dict, List, Optional, Union
 
 from pydantic import BaseModel, Field, constr, validator
@@ -23,8 +22,7 @@
     DisjunctionConfig,
 )
 from .targets import ColumnMatrix, DatasetMatrix
-
-CRON_REGEX = "^(0|\d) (\d+|\d+-\d+|\/\d+)(,\d+|\d+-\d+|\/\d+)* (\*|\d+|\d+-\d+|\/\d+) (\*|\d+|\d+-\d+|\/\d+) (\*|\d+|\d+-\d+|(1|2|3|4|5|6|0)(,\d+)*)$"
+from whylabs_toolkit.helpers.cron_validators import validate_cron_expression
 
 
 class Analyzer(NoExtrasBaseModel):
@@ -108,7 +106,7 @@ def validate_schedule(
         cls, v: Optional[Union[FixedCadenceSchedule, CronSchedule]]
     ) -> Optional[Union[FixedCadenceSchedule, CronSchedule]]:
         """Validate the schedule."""
-        if isinstance(v, CronSchedule) and not re.match(CRON_REGEX, v.cron):
+        if isinstance(v, CronSchedule) and not validate_cron_expression(v.cron):
             raise ValueError("CronSchedule must be no less granular than 1 hour and must have 5 fields.")
         return v