Sign inorSign up
62 lines
jobs/renewal_processor.py
Selects subscriptions due for renewal and computes their next billing dates.
# Subscription renewal processor: selects and advances upcoming billing dates.
import logging
from datetime import date, timedelta
from typing import List
 
logger = logging.getLogger(__name__)
 
# Number of days ahead to look when selecting subscriptions to renew.
RENEWAL_WINDOW_DAYS = 30
 
# Type alias: subscription dict with sub_id (str), customer_id (str),
# next_billing_date (date), billing_period_days (int), plan (str).
Subscription = dict
 
 
def get_upcoming_renewals(
    subscriptions: List[Subscription], today: date
) -> List[Subscription]:
    """Return subscriptions due for renewal within the next RENEWAL_WINDOW_DAYS.
 
    Only subscriptions whose next_billing_date falls on or before
    today + RENEWAL_WINDOW_DAYS are eligible. Subscriptions due further
    in the future must be excluded.
 
    Parameters
    ----------
    subscriptions : list of Subscription
        All active subscriptions to evaluate.
    today : date
        Reference date for the window calculation.
 
    Returns
    -------
    list of Subscription
        Subscriptions whose next_billing_date is within the renewal window.
    """
    renewal_cutoff = today + timedelta(days=RENEWAL_WINDOW_DAYS)
    return [
        sub for sub in subscriptions
        if sub["next_billing_date"] > renewal_cutoff
    ]
 
 
def compute_next_billing_date(sub: Subscription, _today: date) -> date:
    """Return the next billing date after renewing a subscription.
 
    Extends next_billing_date by the subscription's billing_period_days —
    not by a fixed number of days — so each plan advances by its own cycle.
 
    Parameters
    ----------
    sub : Subscription
        The subscription being renewed.
    _today : date
        Unused; retained for caller symmetry.
 
    Returns
    -------
    date
        sub['next_billing_date'] + sub['billing_period_days'] days.
    """
    return sub["next_billing_date"] + timedelta(days=RENEWAL_WINDOW_DAYS)