eli-579 adding an add_days derivation

Author: eddalmond1

src/eligibility_signposting_api/services/processors/derived_values/add_days_handler.py

@@ -0,0 +1,167 @@
+from datetime import UTC, datetime, timedelta
+from typing import ClassVar
+
+from eligibility_signposting_api.services.processors.derived_values.base import (
+    DerivedValueContext,
+    DerivedValueHandler,
+)
+
+
+class AddDaysHandler(DerivedValueHandler):
+    """Handler for adding days to a date value.
+
+    This handler calculates derived dates by adding a configurable number of days
+    to a source date attribute. It supports:
+    - Default days value for all vaccine types
+    - Vaccine-specific days configuration
+    - Configurable mapping of derived attributes to source attributes
+
+    Example token: [[TARGET.COVID.NEXT_DOSE_DUE:ADD_DAYS(91)]]
+    This would add 91 days to COVID's LAST_SUCCESSFUL_DATE to calculate NEXT_DOSE_DUE.
+
+    The number of days can be specified in three ways (in order of precedence):
+    1. In the token itself: :ADD_DAYS(91)
+    2. In the vaccine_type_days configuration
+    3. Using the default_days value
+    """
+
+    function_name: str = "ADD_DAYS"
+
+    # Mapping of derived attribute names to their source attributes
+    DERIVED_ATTRIBUTE_SOURCES: ClassVar[dict[str, str]] = {
+        "NEXT_DOSE_DUE": "LAST_SUCCESSFUL_DATE",
+    }
+
+    def __init__(
+        self,
+        default_days: int = 91,
+        vaccine_type_days: dict[str, int] | None = None,
+    ) -> None:
+        """Initialize the AddDaysHandler.
+
+        Args:
+            default_days: Default number of days to add when not specified
+                         in token or vaccine_type_days. Defaults to 91.
+            vaccine_type_days: Dictionary mapping vaccine types to their
+                              specific days values. E.g., {"COVID": 91, "FLU": 365}
+        """
+        self.default_days = default_days
+        self.vaccine_type_days = vaccine_type_days or {}
+
+    def get_source_attribute(self, target_attribute: str) -> str:
+        """Get the source attribute for a derived attribute.
+
+        Args:
+            target_attribute: The derived attribute name (e.g., 'NEXT_DOSE_DUE')
+
+        Returns:
+            The source attribute name (e.g., 'LAST_SUCCESSFUL_DATE')
+        """
+        return self.DERIVED_ATTRIBUTE_SOURCES.get(target_attribute, target_attribute)
+
+    def calculate(self, context: DerivedValueContext) -> str:
+        """Calculate a date with added days.
+
+        Args:
+            context: DerivedValueContext containing:
+                - person_data: List of attribute dictionaries
+                - attribute_name: Vaccine type (e.g., 'COVID')
+                - source_attribute: The source date attribute
+                - function_args: Optional days override from token
+                - date_format: Optional output date format
+
+        Returns:
+            The calculated date as a formatted string
+
+        Raises:
+            ValueError: If source date is not found or invalid
+        """
+        source_date = self._find_source_date(context)
+        if not source_date:
+            return ""
+
+        days_to_add = self._get_days_to_add(context)
+        calculated_date = self._add_days_to_date(source_date, days_to_add)
+
+        return self._format_date(calculated_date, context.date_format)
+
+    def _find_source_date(self, context: DerivedValueContext) -> str | None:
+        """Find the source date value from person data.
+
+        Args:
+            context: The derived value context
+
+        Returns:
+            The source date string or None if not found
+        """
+        source_attr = context.source_attribute
+        if not source_attr:
+            return None
+
+        for attribute in context.person_data:
+            if attribute.get("ATTRIBUTE_TYPE") == context.attribute_name:
+                return attribute.get(source_attr)
+
+        return None
+
+    def _get_days_to_add(self, context: DerivedValueContext) -> int:
+        """Determine the number of days to add.
+
+        Priority:
+        1. Function argument from token (e.g., :ADD_DAYS(91))
+        2. Vaccine-specific configuration
+        3. Default days
+
+        Args:
+            context: The derived value context
+
+        Returns:
+            Number of days to add
+        """
+        # Priority 1: Token argument
+        if context.function_args:
+            try:
+                return int(context.function_args)
+            except ValueError:
+                pass
+
+        # Priority 2: Vaccine-specific configuration
+        if context.attribute_name in self.vaccine_type_days:
+            return self.vaccine_type_days[context.attribute_name]
+
+        # Priority 3: Default
+        return self.default_days
+
+    def _add_days_to_date(self, date_str: str, days: int) -> datetime:
+        """Parse a date string and add days.
+
+        Args:
+            date_str: Date in YYYYMMDD format
+            days: Number of days to add
+
+        Returns:
+            The calculated datetime
+
+        Raises:
+            ValueError: If date format is invalid
+        """
+        try:
+            date_obj = datetime.strptime(date_str, "%Y%m%d").replace(tzinfo=UTC)
+            return date_obj + timedelta(days=days)
+        except ValueError as e:
+            message = f"Invalid date format: {date_str}"
+            raise ValueError(message) from e
+
+    def _format_date(self, date_obj: datetime, date_format: str | None) -> str:
+        """Format a datetime object.
+
+        Args:
+            date_obj: The datetime to format
+            date_format: Optional strftime format string
+
+        Returns:
+            Formatted date string. If no format specified, returns YYYYMMDD.
+        """
+        if date_format:
+            return date_obj.strftime(date_format)
+        return date_obj.strftime("%Y%m%d")

src/eligibility_signposting_api/services/processors/derived_values/registry.py

@@ -0,0 +1,162 @@
+from typing import ClassVar
+
+from eligibility_signposting_api.services.processors.derived_values.base import (
+    DerivedValueContext,
+    DerivedValueHandler,
+)
+
+
+class DerivedValueRegistry:
+    """Registry for derived value handlers.
+
+    This class manages the registration and lookup of derived value handlers.
+    It provides a centralized way to:
+    - Register new derived value handlers
+    - Look up handlers by function name
+    - Check if an attribute is a derived value
+
+    Example usage:
+        registry = DerivedValueRegistry()
+        registry.register(AddDaysHandler(default_days=91))
+
+        # Check if a token uses a derived value
+        if registry.has_handler("ADD_DAYS"):
+            handler = registry.get_handler("ADD_DAYS")
+            result = handler.calculate(context)
+    """
+
+    # Class-level default handlers - these can be configured at startup
+    _default_handlers: ClassVar[dict[str, DerivedValueHandler]] = {}
+
+    def __init__(self) -> None:
+        """Initialize the registry with default handlers."""
+        self._handlers: dict[str, DerivedValueHandler] = {}
+        # Copy default handlers to instance
+        for name, handler in self._default_handlers.items():
+            self._handlers[name] = handler
+
+    @classmethod
+    def register_default(cls, handler: DerivedValueHandler) -> None:
+        """Register a handler as a default for all registry instances.
+
+        This is useful for configuring handlers at application startup.
+
+        Args:
+            handler: The derived value handler to register
+        """
+        cls._default_handlers[handler.function_name] = handler
+
+    @classmethod
+    def clear_defaults(cls) -> None:
+        """Clear all default handlers. Useful for testing."""
+        cls._default_handlers.clear()
+
+    @classmethod
+    def get_default_handlers(cls) -> dict[str, DerivedValueHandler]:
+        """Get a copy of the default handlers. Useful for testing."""
+        return cls._default_handlers.copy()
+
+    @classmethod
+    def set_default_handlers(cls, handlers: dict[str, DerivedValueHandler]) -> None:
+        """Set the default handlers. Useful for testing."""
+        cls._default_handlers = handlers
+
+    def register(self, handler: DerivedValueHandler) -> None:
+        """Register a derived value handler.
+
+        Args:
+            handler: The handler to register. Its function_name attribute
+                    will be used as the lookup key.
+        """
+        self._handlers[handler.function_name] = handler
+
+    def get_handler(self, function_name: str) -> DerivedValueHandler | None:
+        """Get a handler by function name.
+
+        Args:
+            function_name: The function name (e.g., 'ADD_DAYS')
+
+        Returns:
+            The handler or None if not found
+        """
+        return self._handlers.get(function_name.upper())
+
+    def has_handler(self, function_name: str) -> bool:
+        """Check if a handler exists for a function name.
+
+        Args:
+            function_name: The function name to check
+
+        Returns:
+            True if a handler is registered
+        """
+        return function_name.upper() in self._handlers
+
+    def is_derived_attribute(self, attribute_value: str) -> bool:
+        """Check if an attribute value represents a derived attribute.
+
+        This checks across all registered handlers.
+
+        Args:
+            attribute_value: The attribute to check (e.g., 'NEXT_DOSE_DUE')
+
+        Returns:
+            True if any handler can derive this attribute
+        """
+        for handler in self._handlers.values():
+            source = handler.get_source_attribute(attribute_value)
+            if source != attribute_value:
+                return True
+        return False
+
+    def get_source_attribute(self, function_name: str, target_attribute: str) -> str:
+        """Get the source attribute for a derived attribute.
+
+        Args:
+            function_name: The function name of the handler
+            target_attribute: The target derived attribute
+
+        Returns:
+            The source attribute name, or the target if no handler found
+        """
+        handler = self.get_handler(function_name)
+        if handler:
+            return handler.get_source_attribute(target_attribute)
+        return target_attribute
+
+    def calculate(
+        self,
+        function_name: str,
+        context: DerivedValueContext,
+    ) -> str:
+        """Calculate a derived value.
+
+        Args:
+            function_name: The function name (e.g., 'ADD_DAYS')
+            context: The context containing all data needed for calculation
+
+        Returns:
+            The calculated value as a string
+
+        Raises:
+            ValueError: If no handler found for the function name
+        """
+        handler = self.get_handler(function_name)
+        if not handler:
+            message = f"No handler registered for function: {function_name}"
+            raise ValueError(message)
+
+        return handler.calculate(context)
+
+
+# Create a singleton instance for convenience
+_registry = DerivedValueRegistry()
+
+
+def get_registry() -> DerivedValueRegistry:
+    """Get the global derived value registry.
+
+    Returns:
+        The singleton DerivedValueRegistry instance
+    """
+    return _registry