Components / recurrence

recurrence

iCal RRULE-compliant recurrence calculation engine

v0.1.0 stable queries events hooks

Recurrence provides pattern-based scheduling for recurring entities. It supports daily, weekly, monthly, and yearly patterns with flexible restrictions (specific days, months). Uses python-dateutil for iCal RRULE compliance. Components inject RecurrencePort to calculate next occurrences, and MaterializationPolicy to control when new instances are created.

1
Ports
2
Schemas
2
Hooks
2
Events
01

Ports

Required
  • Clock
    Current time for occurrence calculations
Optional
  • EventBus optional
    Publish RecurrenceTriggered events
Adapters Provided
  • RecurrenceEngine
    implements RecurrencePort
    iCal RRULE-compliant recurrence calculation
02

Schemas

Defines
Uses

No external schemas used

03

Hooks

  • on_recurrence_trigger
    When a recurrence generates a new occurrence
    outputs: Patch, SideEffectRequest
  • before_materialize
    Before materializing a new occurrence
    outputs: Veto, Patch
04

Events

  • RecurrenceTriggered v1
    After a recurrence generates a new occurrence
    payload: {source_id, new_id, spec, triggered_at}
  • RecurrenceEnded v1
    When a recurrence reaches its end condition
    payload: {source_id, spec, ended_at, total_triggers}
05

Stories

Story
Daily Recurrence

Calculate daily occurrences from a start date
Story
Weekly on Mon/Wed/Fri

Weekly recurrence on specific days
Story
Monthly on the 15th

Monthly recurrence on a specific day of month
Story
End After Count

Recurrence that ends after N occurrences
Story
End After Date

Recurrence that ends on a specific date
06

Examples

Calculate next occurrence 
from datetime import datetime, UTC
from psp.platform.recurrence import RecurrenceEngine, RecurrenceSpec, Frequency

engine = RecurrenceEngine()
spec = RecurrenceSpec(freq=Frequency.DAILY)
now = datetime(2024, 6, 15, 12, 0, tzinfo=UTC)

next_date = engine.next_occurrence(spec, after=now, dtstart=now)
print(f"Next: {next_date}")  # 2024-06-16 12:00:00+00:00

Use RecurrenceEngine to find the next occurrence.

Weekly on Mon/Wed/Fri 
from psp.platform.recurrence import (
    RecurrenceEngine, RecurrenceSpec, Frequency, Weekday
)

engine = RecurrenceEngine()
spec = RecurrenceSpec(
    freq=Frequency.WEEKLY,
    byday=(Weekday.MO, Weekday.WE, Weekday.FR),
)

start = datetime(2024, 6, 10, 9, 0, tzinfo=UTC)  # Monday
dates = engine.occurrences(spec, dtstart=start, count=5)

Configure weekly recurrence on specific days.

Monthly on the 15th 
spec = RecurrenceSpec(
    freq=Frequency.MONTHLY,
    bymonthday=(15,),
)

start = datetime(2024, 1, 15, 9, 0, tzinfo=UTC)
dates = engine.occurrences(spec, dtstart=start, count=3)
# Jan 15, Feb 15, Mar 15

Recur on a specific day of each month.

Use in CompleteTask use case 
class CompleteTaskUseCase:
    def __init__(
        self,
        repo: TaskRepository,
        recurrence: RecurrencePort,
        clock: Clock,
    ):
        self._repo = repo
        self._recurrence = recurrence
        self._clock = clock

    def execute(self, task_id: UUID) -> Task:
        task = self._repo.get(task_id)
        now = self._clock.now()

        task.complete(now)

        # If task has recurrence, calculate next occurrence
        if task.recurrence_state and task.recurrence_state.is_active:
            next_date = self._recurrence.next_occurrence(
                task.recurrence_state.spec,
                after=now,
                dtstart=task.recurrence_state.dtstart,
            )
            if next_date:
                next_task = task.clone_for_recurrence(due_at=next_date)
                self._repo.save(next_task)
                task.recurrence_state.increment(now)

        self._repo.save(task)
        return task

Trigger recurrence when task is completed.

API Reference

This component mounts routes under /v1/recurrence. View OpenAPI specification