plpipes.action.driver.loop

RunAsOfDateIterator

Bases: _ListIterator

Iterator for run-as-of-date values based on specified parameters.

Parameters:

Name	Type	Description	Default
key	str	The key for the iterator.	required
icfg	dict	Configuration settings for the iterator.	required

Source code in src\plpipes\action\driver\loop.py

class RunAsOfDateIterator(_ListIterator):
    """
    Iterator for run-as-of-date values based on specified parameters.

    Args:
        key (str): The key for the iterator.
        icfg (dict): Configuration settings for the iterator.
    """

    def __init__(self, key, icfg):
        """
        Initializes the run-as-of-date iterator.

        Args:
            key (str): The key for the iterator.
            icfg (dict): Configuration settings for the iterator.
        """
        values = icfg.get("values")
        icfg.setdefault("target", "run.as_of_date")

        start = icfg.get("start")
        if start is not None:
            end = icfg.get("end", "today")
            periodicity = icfg.get("periodicity", "daily")
            values = self._date_range(start, end, periodicity, values)
        elif values is None:
            raise ValueError("No values or start date provided for RunAsOfDateIterator")

        super().__init__(key, icfg, list(values))

    def _date_range(self, start, end, periodicity, more_values):
        """
        Generates a range of dates based on the specified parameters.

        Args:
            start (str): Start date.
            end (str): End date.
            periodicity (str): The frequency of the dates (daily, weekly, etc.).
            more_values (list): Additional values to include in the range.

        Returns:
            list: A sorted list of dates within the specified range.
        """
        from friendlydateparser import parse_date
        from dateutil.relativedelta import relativedelta
        start = parse_date(start)
        end = parse_date(end)
        if periodicity == "daily":
            step = relativedelta(days=1)
        elif periodicity == "weekly":
            step = relativedelta(days=7)
        elif periodicity == "monthly":
            step = relativedelta(months=1)
        elif periodicity == "yearly":
            step = relativedelta(years=1)
        else:
            raise ValueError(f"Unsupported periodicity {periodicity}")

        if more_values is None:
            more_values = []
        values = set(fdp.parse_date(v) for v in more_values)
        value = start
        while value <= end:
            values.add(value)
            value += step
        values = sorted(values)
        logging.debug(f"Date range from {start} to {end} with periodicity {periodicity} yields {values}")
        return values

    def reset(self):
        """Resets the current date to 'now' and initializes the run date."""
        super().reset()
        cfg['run.as_of_date'] = 'now'
        init_run_as_of_date()

    def next(self):
        """
        Advances the iterator to the next date value.

        Returns:
            bool: True if there is a next date; False otherwise.
        """
        if super().next():
            init_run_as_of_date()
            return True
        return False

__init__(key, icfg)

Initializes the run-as-of-date iterator.

Parameters:

Name	Type	Description	Default
key	str	The key for the iterator.	required
icfg	dict	Configuration settings for the iterator.	required

Source code in src\plpipes\action\driver\loop.py

def __init__(self, key, icfg):
    """
    Initializes the run-as-of-date iterator.

    Args:
        key (str): The key for the iterator.
        icfg (dict): Configuration settings for the iterator.
    """
    values = icfg.get("values")
    icfg.setdefault("target", "run.as_of_date")

    start = icfg.get("start")
    if start is not None:
        end = icfg.get("end", "today")
        periodicity = icfg.get("periodicity", "daily")
        values = self._date_range(start, end, periodicity, values)
    elif values is None:
        raise ValueError("No values or start date provided for RunAsOfDateIterator")

    super().__init__(key, icfg, list(values))

next()

Advances the iterator to the next date value.

Returns:

Name	Type	Description
bool		True if there is a next date; False otherwise.

Source code in src\plpipes\action\driver\loop.py

def next(self):
    """
    Advances the iterator to the next date value.

    Returns:
        bool: True if there is a next date; False otherwise.
    """
    if super().next():
        init_run_as_of_date()
        return True
    return False

reset()

Resets the current date to 'now' and initializes the run date.

Source code in src\plpipes\action\driver\loop.py

def reset(self):
    """Resets the current date to 'now' and initializes the run date."""
    super().reset()
    cfg['run.as_of_date'] = 'now'
    init_run_as_of_date()