Recurring time window filter

featuremanagement/_defaultfilters.py

@@ -9,6 +9,7 @@
 from email.utils import parsedate_to_datetime
 from typing import cast, List, Mapping, Optional, Dict, Any
 from ._featurefilters import FeatureFilter
+from ._time_window_filter import Recurrence, is_match, TimeWindowFilterSettings
 
 FEATURE_FLAG_NAME_KEY = "feature_name"
 ROLLOUT_PERCENTAGE_KEY = "RolloutPercentage"
@@ -18,6 +19,15 @@
 # Time Window Constants
 START_KEY = "Start"
 END_KEY = "End"
+TIME_WINDOW_FILTER_SETTING_RECURRENCE = "Recurrence"
+
+# Time Window Exceptions
+TIME_WINDOW_FILTER_INVALID = (
+    "{}: The {} feature filter is not valid for feature {}. It must specify either {}, {}, or both."
+)
+TIME_WINDOW_FILTER_INVALID_RECURRENCE = (
+    "{}: The {} feature filter is not valid for feature {}. It must specify both {} and {} when Recurrence is not None."
+)
 
 # Targeting kwargs
 TARGETED_USER_KEY = "user"
@@ -31,6 +41,8 @@
 FEATURE_FILTER_NAME_KEY = "Name"
 IGNORE_CASE_KEY = "ignore_case"
 
+logger = logging.getLogger(__name__)
+
 
 class TargetingException(Exception):
     """
@@ -52,19 +64,35 @@ def evaluate(self, context: Mapping[Any, Any], **kwargs: Any) -> bool:
         :return: True if the current time is within the time window.
         :rtype: bool
         """
-        start = context.get(PARAMETERS_KEY, {}).get(START_KEY)
-        end = context.get(PARAMETERS_KEY, {}).get(END_KEY)
+        start = context.get(PARAMETERS_KEY, {}).get(START_KEY, None)
+        end = context.get(PARAMETERS_KEY, {}).get(END_KEY, None)
+        recurrence_data = context.get(PARAMETERS_KEY, {}).get(TIME_WINDOW_FILTER_SETTING_RECURRENCE, None)
+        recurrence = None
 
         current_time = datetime.now(timezone.utc)
 
         if not start and not end:
-            logging.warning("%s: At least one of Start or End is required.", TimeWindowFilter.__name__)
+            logger.warning(
+                TIME_WINDOW_FILTER_INVALID,
+                TimeWindowFilter.__name__,
+                context.get(FEATURE_FLAG_NAME_KEY),
+                START_KEY,
+                END_KEY,
+            )
             return False
 
-        start_time = parsedate_to_datetime(start) if start else None
-        end_time = parsedate_to_datetime(end) if end else None
+        start_time: Optional[datetime] = parsedate_to_datetime(start) if start else None
+        end_time: Optional[datetime] = parsedate_to_datetime(end) if end else None
+
+        if (start_time is None or start_time <= current_time) and (end_time is None or current_time < end_time):
+            return True
+
+        if recurrence_data:
+            recurrence = Recurrence(recurrence_data)
+            settings = TimeWindowFilterSettings(start_time, end_time, recurrence)
+            return is_match(settings, current_time)
 
-        return (start_time is None or start_time <= current_time) and (end_time is None or current_time < end_time)
+        return False
 
 
 @FeatureFilter.alias("Microsoft.Targeting")

featuremanagement/_time_window_filter/__init__.py

@@ -0,0 +1,9 @@
+# ------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+from ._recurrence_evaluator import is_match
+from ._models import Recurrence, TimeWindowFilterSettings
+
+__all__ = ["is_match", "Recurrence", "TimeWindowFilterSettings"]

featuremanagement/_time_window_filter/_models.py

@@ -0,0 +1,149 @@
+# ------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+from enum import Enum
+from typing import Dict, Any, Optional, List
+from datetime import datetime
+from dataclasses import dataclass
+from email.utils import parsedate_to_datetime
+
+
+class RecurrencePatternType(str, Enum):
+    """
+    The recurrence pattern type.
+    """
+
+    DAILY = "Daily"
+    WEEKLY = "Weekly"
+
+    @staticmethod
+    def from_str(value: str) -> "RecurrencePatternType":
+        """
+        Get the RecurrencePatternType from the string value.
+
+        :param value: The string value.
+        :type value: str
+        :return: The RecurrencePatternType.
+        :rtype: RecurrencePatternType
+        """
+        if value == "Daily":
+            return RecurrencePatternType.DAILY
+        if value == "Weekly":
+            return RecurrencePatternType.WEEKLY
+        raise ValueError(f"Invalid value: {value}")
+
+
+class RecurrenceRangeType(str, Enum):
+    """
+    The recurrence range type.
+    """
+
+    NO_END = "NoEnd"
+    END_DATE = "EndDate"
+    NUMBERED = "Numbered"
+
+    @staticmethod
+    def from_str(value: str) -> "RecurrenceRangeType":
+        """
+        Get the RecurrenceRangeType from the string value.
+
+        :param value: The string value.
+        :type value: str
+        :return: The RecurrenceRangeType.
+        :rtype: RecurrenceRangeType
+        """
+        if value == "NoEnd":
+            return RecurrenceRangeType.NO_END
+        if value == "EndDate":
+            return RecurrenceRangeType.END_DATE
+        if value == "Numbered":
+            return RecurrenceRangeType.NUMBERED
+        raise ValueError(f"Invalid value: {value}")
+
+
+class RecurrencePattern:  # pylint: disable=too-few-public-methods
+    """
+    The recurrence pattern settings.
+    """
+
+    days: List[str] = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
+
+    def __init__(self, pattern_data: Dict[str, Any]):
+        self.type = RecurrencePatternType.from_str(pattern_data.get("Type", "Daily"))
+        self.interval = pattern_data.get("Interval", 1)
+        if self.interval <= 0:
+            raise ValueError("The interval must be greater than 0.")
+        # Days of the week are represented as a list of strings of their names.
+        days_of_week_str = pattern_data.get("DaysOfWeek", [])
+
+        # Days of the week are represented as a list of integers from 0 to 6.
+        self.days_of_week: List[int] = []
+        for day in days_of_week_str:
+            if day not in self.days:
+                raise ValueError(f"Invalid value for DaysOfWeek: {day}")
+            if self.days.index(day) in self.days_of_week:
+                raise ValueError(f"Duplicate day of the week found: {day}")
+            self.days_of_week.append(self.days.index(day))
+        if pattern_data.get("FirstDayOfWeek") and pattern_data.get("FirstDayOfWeek") not in self.days:
+            raise ValueError(f"Invalid value for FirstDayOfWeek: {pattern_data.get('FirstDayOfWeek')}")
+        self.first_day_of_week = self.days.index(pattern_data.get("FirstDayOfWeek", "Sunday"))
+
+
+class RecurrenceRange:  # pylint: disable=too-few-public-methods
+    """
+    The recurrence range settings.
+    """
+
+    type: RecurrenceRangeType
+    end_date: Optional[datetime] = None
+
+    def __init__(self, range_data: Dict[str, Any]):
+        self.type = RecurrenceRangeType.from_str(range_data.get("Type", "NoEnd"))
+        if range_data.get("EndDate") and isinstance(range_data.get("EndDate"), str):
+            end_date_str = range_data.get("EndDate", "")
+            try:
+                self.end_date = parsedate_to_datetime(end_date_str) if end_date_str else None
+            except ValueError as e:
+                raise ValueError(f"Invalid value for EndDate: {end_date_str}") from e
+            except TypeError as e:
+                # Python 3.9 and earlier throw TypeError if the string is not in RFC 2822 format.
+                raise ValueError(f"Invalid value for EndDate: {end_date_str}") from e
+        self.num_of_occurrences = range_data.get("NumberOfOccurrences", 2**63 - 1)
+        if self.num_of_occurrences <= 0:
+            raise ValueError("The number of occurrences must be greater than 0.")
+
+
+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", {}))
+
+
+@dataclass
+class TimeWindowFilterSettings:
+    """
+    The settings for the time window filter.
+    """
+
+    start: Optional[datetime]
+    end: Optional[datetime]
+    recurrence: Optional[Recurrence]
+
+
+@dataclass
+class OccurrenceInfo:
+    """
+    The information of the previous occurrence.
+    """
+
+    previous_occurrence: datetime
+    num_of_occurrences: int

featuremanagement/_time_window_filter/_recurrence_evaluator.py

@@ -0,0 +1,125 @@
+# ------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for
+# license information.
+# -------------------------------------------------------------------------
+from datetime import datetime, timedelta
+from typing import Optional
+from ._models import RecurrencePatternType, RecurrenceRangeType, TimeWindowFilterSettings, OccurrenceInfo, Recurrence
+from ._recurrence_validator import validate_settings, _get_passed_week_days, _sort_days_of_week
+
+DAYS_PER_WEEK = 7
+REQUIRED_PARAMETER = "Required parameter: %s"
+
+
+def is_match(settings: TimeWindowFilterSettings, now: datetime) -> bool:
+    """
+    Check if the current time is within the time window filter settings.
+
+    :param TimeWindowFilterSettings settings: The settings for the time window filter.
+    :param datetime now: The current time.
+    :return: True if the current time is within the time window filter settings, otherwise False.
+    :rtype: bool
+    """
+    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")
+
+    validate_settings(recurrence, start, end)
+
+    previous_occurrence = _get_previous_occurrence(recurrence, start, now)
+    if previous_occurrence is None:
+        return False
+
+    occurrence_end_date = previous_occurrence + (end - start)
+    return now < occurrence_end_date
+
+
+def _get_previous_occurrence(recurrence: Recurrence, start: datetime, now: datetime) -> Optional[datetime]:
+    if now < start:
+        return None
+
+    pattern_type = recurrence.pattern.type
+    if pattern_type == RecurrencePatternType.DAILY:
+        occurrence_info = _get_daily_previous_occurrence(recurrence, start, now)
+    elif pattern_type == RecurrencePatternType.WEEKLY:
+        occurrence_info = _get_weekly_previous_occurrence(recurrence, start, now)
+    else:
+        raise ValueError(f"Unsupported recurrence pattern type: {pattern_type}")
+
+    recurrence_range = recurrence.range
+    range_type = recurrence_range.type
+    previous_occurrence = occurrence_info.previous_occurrence
+    end_date = recurrence_range.end_date
+    if (
+        range_type == RecurrenceRangeType.END_DATE
+        and previous_occurrence is not None
+        and end_date is not None
+        and previous_occurrence > end_date
+    ):
+        return None
+    if (
+        range_type == RecurrenceRangeType.NUMBERED
+        and recurrence_range.num_of_occurrences is not None
+        and occurrence_info.num_of_occurrences > recurrence_range.num_of_occurrences
+    ):
+        return None
+
+    return occurrence_info.previous_occurrence
+
+
+def _get_daily_previous_occurrence(recurrence: Recurrence, start: datetime, now: datetime) -> OccurrenceInfo:
+    interval = recurrence.pattern.interval
+    num_of_occurrences = (now - start).days // interval
+    previous_occurrence = start + timedelta(days=num_of_occurrences * interval)
+    return OccurrenceInfo(previous_occurrence, num_of_occurrences + 1)
+
+
+def _get_weekly_previous_occurrence(recurrence: Recurrence, start: datetime, now: datetime) -> OccurrenceInfo:
+    pattern = recurrence.pattern
+    interval = pattern.interval
+    first_day_of_first_week = start - timedelta(days=_get_passed_week_days(start.weekday(), pattern.first_day_of_week))
+
+    number_of_interval = (now - first_day_of_first_week).days // (interval * DAYS_PER_WEEK)
+    first_day_of_most_recent_occurring_week = first_day_of_first_week + timedelta(
+        days=number_of_interval * (interval * DAYS_PER_WEEK)
+    )
+    sorted_days_of_week = _sort_days_of_week(pattern.days_of_week, pattern.first_day_of_week)
+    max_day_offset = _get_passed_week_days(sorted_days_of_week[-1], pattern.first_day_of_week)
+    min_day_offset = _get_passed_week_days(sorted_days_of_week[0], pattern.first_day_of_week)
+    num_of_occurrences = number_of_interval * len(sorted_days_of_week) - sorted_days_of_week.index(start.weekday())
+
+    if now > first_day_of_most_recent_occurring_week + timedelta(days=DAYS_PER_WEEK):
+        num_of_occurrences += len(sorted_days_of_week)
+        most_recent_occurrence = first_day_of_most_recent_occurring_week + timedelta(days=max_day_offset)
+        return OccurrenceInfo(most_recent_occurrence, num_of_occurrences)
+
+    day_with_min_offset = first_day_of_most_recent_occurring_week + timedelta(days=min_day_offset)
+    if start > day_with_min_offset:
+        num_of_occurrences = 0
+        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)
+        )
+    else:
+        most_recent_occurrence = day_with_min_offset
+        num_of_occurrences += 1
+
+        for day in sorted_days_of_week[sorted_days_of_week.index(day_with_min_offset.weekday()) + 1 :]:
+            day_with_min_offset = first_day_of_most_recent_occurring_week + timedelta(
+                days=_get_passed_week_days(day, pattern.first_day_of_week)
+            )
+            if now < day_with_min_offset:
+                break
+            most_recent_occurrence = day_with_min_offset
+            num_of_occurrences += 1
+
+    return OccurrenceInfo(most_recent_occurrence, num_of_occurrences)