Updating to deal with optional start/end

Author: mrm9084

featuremanagement/_time_window_filter/_models.py

@@ -5,7 +5,7 @@
 # -------------------------------------------------------------------------
 import math
 from enum import Enum
-from typing import Dict, Any, Optional
+from typing import Dict, Any, Optional, List
 from datetime import datetime
 from dataclasses import dataclass
 from email.utils import parsedate_to_datetime
@@ -69,8 +69,8 @@ class RecurrencePattern:  # pylint: disable=too-few-public-methods
     The recurrence pattern settings.
     """
 
-    days: list = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
-    days_of_week: list = []
+    days: List[str] = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
+    days_of_week: List[int] = []
 
     def __init__(self, pattern_data: Dict[str, Any]):
         self.type = RecurrencePatternType.from_str(pattern_data.get("Type", "Daily"))
@@ -106,6 +106,9 @@ class Recurrence:  # pylint: disable=too-few-public-methods
     The recurrence settings.
     """
 
+    pattern: RecurrencePattern
+    range: RecurrenceRange
+
     def __init__(self, recurrence_data: Dict[str, Any]):
         self.pattern = RecurrencePattern(recurrence_data.get("Pattern", {}))
         self.range = RecurrenceRange(recurrence_data.get("Range", {}))
@@ -117,8 +120,8 @@ class TimeWindowFilterSettings:
     The settings for the time window filter.
     """
 
-    start: datetime
-    end: datetime
+    start: Optional[datetime]
+    end: Optional[datetime]
     recurrence: Optional[Recurrence]
 
 

featuremanagement/_time_window_filter/_recurrence_evaluator.py

@@ -4,12 +4,13 @@
 # license information.
 # -------------------------------------------------------------------------
 from datetime import datetime, timedelta
-from typing import List, Optional
+from typing import Optional
 from ._models import RecurrencePatternType, RecurrenceRangeType, TimeWindowFilterSettings, OccurrenceInfo
-from ._recurrence_validator import validate_settings
+from ._recurrence_validator import validate_settings, _get_passed_week_days, _sort_days_of_week
 
 DAYS_PER_WEEK = 7
 
+
 def is_match(settings: TimeWindowFilterSettings, now: datetime) -> bool:
     """
     Check if the current time is within the time window filter settings.
@@ -40,7 +41,7 @@ def _get_previous_occurrence(settings: TimeWindowFilterSettings, now: datetime)
     elif pattern_type == RecurrencePatternType.WEEKLY:
         occurrence_info = _get_weekly_previous_occurrence(settings, now)
     else:
-        raise ValueError(f"Invalid recurrence pattern type: %s", pattern_type)
+        raise ValueError(f"Invalid recurrence pattern type: {pattern_type}")
 
     recurrence_range = settings.recurrence.range
     range_type = recurrence_range.type
@@ -97,7 +98,9 @@ def _get_weekly_previous_occurrence(settings: TimeWindowFilterSettings, now: dat
         day_with_min_offset = start
     if now < day_with_min_offset:
         most_recent_occurrence = (
-            first_day_of_most_recent_occurring_week - timedelta(days=interval * DAYS_PER_WEEK) + timedelta(days=max_day_offset)
+            first_day_of_most_recent_occurring_week
+            - timedelta(days=interval * DAYS_PER_WEEK)
+            + timedelta(days=max_day_offset)
         )
     else:
         most_recent_occurrence = day_with_min_offset
@@ -113,19 +116,3 @@ def _get_weekly_previous_occurrence(settings: TimeWindowFilterSettings, now: dat
             num_of_occurrences += 1
 
     return OccurrenceInfo(most_recent_occurrence, num_of_occurrences)
-
-
-def _get_passed_week_days(current_day: int, first_day_of_week: int) -> int:
-    """
-    Get the number of days passed since the first day of the week.
-    :param int current_day: The current day of the week (0-6).
-    :param int first_day_of_week: The first day of the week (0-6).
-    :return: The number of days passed since the first day of the week.
-    :rtype: int
-    """
-    return (current_day - first_day_of_week + DAYS_PER_WEEK) % DAYS_PER_WEEK
-
-
-def _sort_days_of_week(days_of_week: List[int], first_day_of_week: int) -> List[int]:
-    sorted_days = sorted(days_of_week)
-    return sorted_days[sorted_days.index(first_day_of_week) :] + sorted_days[: sorted_days.index(first_day_of_week)]

featuremanagement/_time_window_filter/_recurrence_validator.py

@@ -4,8 +4,8 @@
 # license information.
 # -------------------------------------------------------------------------
 from datetime import datetime, timedelta
-from ._models import RecurrencePatternType, RecurrenceRangeType, TimeWindowFilterSettings
-from ._recurrence_evaluator import _get_passed_week_days, _sort_days_of_week
+from typing import List
+from ._models import RecurrencePatternType, RecurrenceRangeType, TimeWindowFilterSettings, Recurrence
 
 
 DAYS_PER_WEEK = 7
@@ -27,13 +27,21 @@ def validate_settings(settings: TimeWindowFilterSettings) -> None:
     :param TimeWindowFilterSettings settings: The settings for the time window filter.
     :raises ValueError: If the settings are invalid.
     """
-    _validate_recurrence_required_parameter(settings)
-    _validate_recurrence_pattern(settings)
-    _validate_recurrence_range(settings)
+    recurrence = settings.recurrence
+    if recurrence is None:
+        raise ValueError(REQUIRED_PARAMETER % "Recurrence")
 
+    start = settings.start
+    end = settings.end
+    if start is None or end is None:
+        raise ValueError(REQUIRED_PARAMETER % "Start or End")
 
-def _validate_recurrence_required_parameter(settings: TimeWindowFilterSettings) -> None:
-    recurrence = settings.recurrence
+    _validate_recurrence_required_parameter(recurrence, start, end)
+    _validate_recurrence_pattern(recurrence, start, end)
+    _validate_recurrence_range(recurrence, start)
+
+
+def _validate_recurrence_required_parameter(recurrence: Recurrence, start: datetime, end: datetime) -> None:
     param_name = ""
     reason = ""
     if recurrence.pattern is None:
@@ -42,86 +50,88 @@ def _validate_recurrence_required_parameter(settings: TimeWindowFilterSettings)
     if recurrence.range is None:
         param_name = f"{RECURRENCE_RANGE}"
         reason = REQUIRED_PARAMETER
-    if not settings.end > settings.start:
+    if not end > start:
         param_name = "end"
         reason = OUT_OF_RANGE
-    if settings.end > settings.start + timedelta(days=TEN_YEARS):
+    if end > start + timedelta(days=TEN_YEARS):
         param_name = "end"
         reason = TIME_WINDOW_DURATION_TEN_YEARS
 
     if param_name:
         raise ValueError(reason % param_name)
 
 
-def _validate_recurrence_pattern(settings: TimeWindowFilterSettings) -> None:
-    pattern_type = settings.recurrence.pattern.type
+def _validate_recurrence_pattern(recurrence: Recurrence, start: datetime, end: datetime) -> None:
+    if recurrence is None:
+        raise ValueError(REQUIRED_PARAMETER % "Recurrence")
+    pattern_type = recurrence.pattern.type
 
     if pattern_type == RecurrencePatternType.DAILY:
-        _validate_daily_recurrence_pattern(settings)
+        _validate_daily_recurrence_pattern(recurrence, start, end)
     else:
-        _validate_weekly_recurrence_pattern(settings)
+        _validate_weekly_recurrence_pattern(recurrence, start, end)
 
 
-def _validate_recurrence_range(settings: TimeWindowFilterSettings) -> None:
-    range_type = settings.recurrence.range.type
+def _validate_recurrence_range(recurrence: Recurrence, start: datetime) -> None:
+    range_type = recurrence.range.type
     if range_type == RecurrenceRangeType.END_DATE:
-        _validate_end_date(settings)
+        _validate_end_date(recurrence, start)
 
 
-def _validate_daily_recurrence_pattern(settings: TimeWindowFilterSettings) -> None:
+def _validate_daily_recurrence_pattern(recurrence: Recurrence, start: datetime, end: datetime) -> None:
     # "Start" is always a valid first occurrence for "Daily" pattern.
     # Only need to check if time window validated
-    _validate_time_window_duration(settings)
+    _validate_time_window_duration(recurrence, start, end)
 
 
-def _validate_weekly_recurrence_pattern(settings: TimeWindowFilterSettings) -> None:
-    _validate_days_of_week(settings)
+def _validate_weekly_recurrence_pattern(recurrence: Recurrence, start: datetime, end: datetime) -> None:
+    _validate_days_of_week(recurrence)
 
     # Check whether "Start" is a valid first occurrence
-    pattern = settings.recurrence.pattern
-    if settings.start.weekday() not in pattern.days_of_week:
+    pattern = recurrence.pattern
+    if start.weekday() not in pattern.days_of_week:
         raise ValueError(NOT_MATCHED % "start")
 
     # Time window duration must be shorter than how frequently it occurs
-    _validate_time_window_duration(settings)
+    _validate_time_window_duration(recurrence, start, end)
 
     # Check whether the time window duration is shorter than the minimum gap between days of week
-    if not _is_duration_compliant_with_days_of_week(settings):
+    if not _is_duration_compliant_with_days_of_week(recurrence, start, end):
         raise ValueError(TIME_WINDOW_DURATION_OUT_OF_RANGE % "Recurrence.Pattern.DaysOfWeek")
 
 
-def _validate_time_window_duration(settings: TimeWindowFilterSettings) -> None:
-    pattern = settings.recurrence.pattern
+def _validate_time_window_duration(recurrence: Recurrence, start: datetime, end: datetime) -> None:
+    pattern = recurrence.pattern
     interval_duration = (
         timedelta(days=pattern.interval)
         if pattern.type == RecurrencePatternType.DAILY
         else timedelta(days=pattern.interval * DAYS_PER_WEEK)
     )
-    time_window_duration = settings.end - settings.start
+    time_window_duration = end - start
     if time_window_duration > interval_duration:
         raise ValueError(TIME_WINDOW_DURATION_OUT_OF_RANGE % "Recurrence.Pattern.Interval")
 
 
-def _validate_days_of_week(settings: TimeWindowFilterSettings) -> None:
-    days_of_week = settings.recurrence.pattern.days_of_week
+def _validate_days_of_week(recurrence: Recurrence) -> None:
+    days_of_week = recurrence.pattern.days_of_week
     if not days_of_week:
         raise ValueError(REQUIRED_PARAMETER % "Recurrence.Pattern.DaysOfWeek")
 
 
-def _validate_end_date(settings: TimeWindowFilterSettings) -> None:
-    end_date = settings.recurrence.range.end_date
-    if end_date and end_date < settings.start:
+def _validate_end_date(recurrence: Recurrence, start: datetime) -> None:
+    end_date = recurrence.range.end_date
+    if end_date and end_date < start:
         raise ValueError("The Recurrence.Range.EndDate should be after the Start")
 
 
-def _is_duration_compliant_with_days_of_week(settings: TimeWindowFilterSettings) -> bool:
-    days_of_week = settings.recurrence.pattern.days_of_week
+def _is_duration_compliant_with_days_of_week(recurrence: Recurrence, start: datetime, end: datetime) -> bool:
+    days_of_week = recurrence.pattern.days_of_week
     if len(days_of_week) == 1:
         return True
 
     # Get the date of first day of the week
     today = datetime.now()
-    first_day_of_week = settings.recurrence.pattern.first_day_of_week
+    first_day_of_week = recurrence.pattern.first_day_of_week
     offset = _get_passed_week_days(today.weekday(), first_day_of_week)
     first_date_of_week = today - timedelta(days=offset)
     sorted_days_of_week = _sort_days_of_week(days_of_week, first_day_of_week)
@@ -137,7 +147,7 @@ def _is_duration_compliant_with_days_of_week(settings: TimeWindowFilterSettings)
             min_gap = min(min_gap, current_gap)
         prev_occurrence = date
 
-    if settings.recurrence.pattern.interval == 1:
+    if recurrence.pattern.interval == 1:
         # It may cross weeks. Check the adjacent week
         date = first_date_of_week + timedelta(
             days=DAYS_PER_WEEK + _get_passed_week_days(sorted_days_of_week[0], first_day_of_week)
@@ -149,5 +159,21 @@ def _is_duration_compliant_with_days_of_week(settings: TimeWindowFilterSettings)
         current_gap = date - prev_occurrence
         min_gap = min(min_gap, current_gap)
 
-    time_window_duration = settings.end - settings.start
+    time_window_duration = end - start
     return min_gap >= time_window_duration
+
+
+def _get_passed_week_days(current_day: int, first_day_of_week: int) -> int:
+    """
+    Get the number of days passed since the first day of the week.
+    :param int current_day: The current day of the week, where Monday == 0 ... Sunday == 6.
+    :param int first_day_of_week: The first day of the week (0-6), where Monday == 0 ... Sunday == 6.
+    :return: The number of days passed since the first day of the week.
+    :rtype: int
+    """
+    return (current_day - first_day_of_week + DAYS_PER_WEEK) % DAYS_PER_WEEK
+
+
+def _sort_days_of_week(days_of_week: List[int], first_day_of_week: int) -> List[int]:
+    sorted_days = sorted(days_of_week)
+    return sorted_days[sorted_days.index(first_day_of_week) :] + sorted_days[: sorted_days.index(first_day_of_week)]