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,17 +64,45 @@ 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 recurrence_data:
+            recurrence = Recurrence(recurrence_data)
+
         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 recurrence:
+            if start_time and end_time:
+                settings = TimeWindowFilterSettings(start_time, end_time, recurrence)
+                return is_match(settings, current_time)
+            logger.warning(
+                TIME_WINDOW_FILTER_INVALID_RECURRENCE,
+                TimeWindowFilter.__name__,
+                context.get(FEATURE_FLAG_NAME_KEY),
+                START_KEY,
+                END_KEY,
+            )
+            return False
+
+        if not start and not end:
+            logging.warning("%s: At least one of Start or End is required.", TimeWindowFilter.__name__)
+            return False
 
         return (start_time is None or start_time <= current_time) and (end_time is None or current_time < end_time)
 

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,128 @@
+# ------------------------------------------------------------------------
+# 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
+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 = ["Sunday", "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday"]
+
+    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_week = pattern_data.get("DaysOfWeek", [])
+        for day in days_of_week:
+            self.days_of_week.append(self.days.index(day))
+        first_day_of_week = pattern_data.get("FirstDayOfWeek", "Sunday")
+        self.first_day_of_week = self.days.index(first_day_of_week) if first_day_of_week in self.days else 0
+
+
+class RecurrenceRange:  # pylint: disable=too-few-public-methods
+    """
+    The recurrence range settings.
+    """
+
+    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", "")
+            self.end_date = parsedate_to_datetime(end_date_str) if end_date_str else None
+        self.num_of_occurrences = range_data.get("NumberOfOccurrences", 0)
+        if self.num_of_occurrences < 0:
+            raise ValueError("The number of occurrences must be greater than or equal to 0.")
+
+
+class Recurrence:  # pylint: disable=too-few-public-methods
+    """
+    The recurrence settings.
+    """
+
+    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: datetime
+    end: 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,121 @@
+# ------------------------------------------------------------------------
+# 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 List, Optional
+from ._models import RecurrencePatternType, RecurrenceRangeType, TimeWindowFilterSettings, OccurrenceInfo
+from ._recurrence_validator import validate_settings
+
+
+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
+    """
+    validate_settings(settings)
+
+    previous_occurrence = _get_previous_occurrence(settings, now)
+    if previous_occurrence is None:
+        return False
+
+    occurrence_end_date = previous_occurrence + (settings.end - settings.start)
+    return now < occurrence_end_date
+
+
+def _get_previous_occurrence(settings: TimeWindowFilterSettings, now: datetime) -> Optional[datetime]:
+    start = settings.start
+    if now < start:
+        return None
+
+    pattern_type = settings.recurrence.pattern.type
+    if pattern_type == RecurrencePatternType.DAILY:
+        occurrence_info = _get_daily_previous_occurrence(settings, now)
+    else:
+        occurrence_info = _get_weekly_previous_occurrence(settings, now)
+
+    recurrence_range = settings.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(settings: TimeWindowFilterSettings, now: datetime) -> OccurrenceInfo:
+    start = settings.start
+    interval = settings.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(settings: TimeWindowFilterSettings, now: datetime) -> OccurrenceInfo:
+    pattern = settings.recurrence.pattern
+    interval = pattern.interval
+    start = settings.start
+    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 * 7)
+    first_day_of_most_recent_occurring_week = first_day_of_first_week + timedelta(
+        days=number_of_interval * (interval * 7)
+    )
+    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=7):
+        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 * 7) + 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)
+
+
+def _get_passed_week_days(current_day: int, first_day_of_week: int) -> int:
+    return (current_day - first_day_of_week + 7) % 7
+
+
+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)]