tessl/pypi-croniter

croniter provides iteration for datetime object with cron like format

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Core Iteration

Name: tessl/pypi-croniter
Author: tessl

The main croniter class provides datetime iteration capabilities based on cron expressions. It supports both forward and backward iteration, timezone handling, various cron formats (5, 6, and 7 fields), and advanced features like hashed and random expressions.

Capabilities

Croniter Class

The core class that parses cron expressions and provides iteration functionality.

class croniter:
    def __init__(
        self,
        expr_format: str,
        start_time=None,
        ret_type=float,
        day_or=True,
        max_years_between_matches=None,
        is_prev=False,
        hash_id=None,
        implement_cron_bug=False,
        second_at_beginning=None,
        expand_from_start_time=False,
    ):
        """
        Initialize croniter instance.
        
        Parameters:
        - expr_format: Cron expression string (e.g., '0 0 * * *', '*/5 * * * *')
        - start_time: Starting datetime (datetime object, timestamp, or None for current time) 
        - ret_type: Return type for get_next/get_prev (float for timestamp, datetime for datetime object)
        - day_or: How to handle day and day_of_week fields (True=OR logic, False=AND logic)
        - max_years_between_matches: Limit CPU cycles for sparse expressions (default 50)
        - is_prev: Whether iteration goes backward (default False)
        - hash_id: Hash ID for Jenkins-style hashed expressions (bytes or str)
        - implement_cron_bug: Enable bug compatibility mode (default False)
        - second_at_beginning: Put seconds field at beginning instead of end
        - expand_from_start_time: Calculate intervals from start_time instead of calendar
        """

Forward Iteration

Get next matching datetime values.

def get_next(self, ret_type=None, start_time=None, update_current=True):
    """
    Get the next datetime matching the cron expression.
    
    Parameters:
    - ret_type: Override return type (float, datetime, or None for instance default)
    - start_time: Override start time for this calculation
    - update_current: Whether to update internal current time (default True)
    
    Returns:
    Next matching datetime as specified ret_type
    """

def all_next(self, ret_type=None, start_time=None, update_current=None):
    """
    Generator yielding consecutive next matching datetimes.
    
    Parameters:
    - ret_type: Return type for each yielded value
    - start_time: Override start time
    - update_current: Whether to update current time for each iteration
    
    Yields:
    Consecutive next matching datetimes
    """

Backward Iteration

Get previous matching datetime values.

def get_prev(self, ret_type=None, start_time=None, update_current=True):
    """
    Get the previous datetime matching the cron expression.
    
    Parameters:
    - ret_type: Override return type (float, datetime, or None for instance default)
    - start_time: Override start time for this calculation  
    - update_current: Whether to update internal current time (default True)
    
    Returns:
    Previous matching datetime as specified ret_type
    """

def all_prev(self, ret_type=None, start_time=None, update_current=None):
    """
    Generator yielding consecutive previous matching datetimes.
    
    Parameters:
    - ret_type: Return type for each yielded value
    - start_time: Override start time
    - update_current: Whether to update current time for each iteration
    
    Yields:
    Consecutive previous matching datetimes
    """

State Management

Manage the current datetime state of the iterator.

def get_current(self, ret_type=None):
    """
    Get the current datetime without advancing iteration.
    
    Parameters:
    - ret_type: Return type (float, datetime, or None for instance default)
    
    Returns:
    Current datetime as specified ret_type
    """

def set_current(self, start_time, force=True):
    """
    Set the current datetime for iteration.
    
    Parameters:
    - start_time: New current time (datetime object or timestamp)
    - force: Whether to force update even if current time is already set
    
    Returns:
    Updated current timestamp
    """

Iterator Protocol

Croniter implements Python's iterator protocol for easy iteration.

def __iter__(self):
    """Return self as iterator"""

def __next__(self):
    """Get next value (forward or backward based on is_prev flag)"""
    
def iter(self, *args, **kwargs):
    """Return appropriate iterator (all_next or all_prev based on is_prev flag)"""

Utility Methods

Static methods for datetime conversion and calculation.

@staticmethod
def datetime_to_timestamp(d):
    """
    Convert datetime object to UNIX timestamp.
    
    Parameters:
    - d: datetime object
    
    Returns:
    UNIX timestamp as float
    """

@staticmethod
def timedelta_to_seconds(td):
    """
    Convert timedelta object to seconds.
    
    Parameters:
    - td: timedelta object
    
    Returns:
    Total seconds as float
    """

def timestamp_to_datetime(self, timestamp, tzinfo=MARKER):
    """
    Convert UNIX timestamp to datetime object.
    
    Parameters:
    - timestamp: UNIX timestamp
    - tzinfo: Timezone info (uses MARKER sentinel by default, falls back to instance timezone)
    
    Returns:
    datetime object with proper timezone
    """

Expression Expansion

Class method for expanding and normalizing cron expressions.

@classmethod
def expand(
    cls,
    expr_format,
    hash_id=None,
    second_at_beginning=False,
    from_timestamp=None,
):
    """
    Expand cron expression into normalized form with field-by-field breakdown.
    
    Parameters:
    - expr_format: Cron expression string to expand
    - hash_id: Hash ID for hashed expressions (Jenkins-style "H")
    - second_at_beginning: Put seconds field at beginning instead of end
    - from_timestamp: Reference timestamp for expansion calculations
    
    Returns:
    tuple: (expanded_expression_list, field_count)
    
    The expanded expression is a list where each element corresponds to 
    a cron field with all possible values explicitly listed.
    """

Static Utility Methods

Additional static methods for datetime calculations and validation.

@staticmethod
def is_leap(year):
    """
    Check if a year is a leap year.
    
    Parameters:
    - year: Year to check (int)
    
    Returns:
    bool: True if leap year, False otherwise
    """

@classmethod  
def value_alias(cls, val, field_index, len_expressions=UNIX_CRON_LEN):
    """
    Convert field value aliases to numeric values.
    
    Parameters:
    - val: Value to convert (may be numeric or text alias like 'jan', 'mon')
    - field_index: Which cron field this value belongs to
    - len_expressions: Length of cron expression (5, 6, or 7 fields)
    
    Returns:
    Converted numeric value
    """

Usage Examples

Basic Forward Iteration

from croniter import croniter
from datetime import datetime

# Every 15 minutes
base = datetime(2010, 1, 25, 4, 46)
iter = croniter('*/15 * * * *', base)

for i in range(5):
    print(iter.get_next(datetime))
# 2010-01-25 05:00:00
# 2010-01-25 05:15:00  
# 2010-01-25 05:30:00
# 2010-01-25 05:45:00
# 2010-01-25 06:00:00

Backward Iteration

# Get previous matches
base = datetime(2010, 8, 25)
iter = croniter('0 0 1 * *', base)  # First day of each month

print(iter.get_prev(datetime))  # 2010-08-01 00:00:00
print(iter.get_prev(datetime))  # 2010-07-01 00:00:00
print(iter.get_prev(datetime))  # 2010-06-01 00:00:00

Generator Usage

# Using generators for continuous iteration
iter = croniter('0 */6 * * *', datetime(2010, 1, 1))  # Every 6 hours

# Get first 10 matches
matches = [next(iter.all_next(datetime)) for _ in range(10)]

# Or use as iterator directly
for i, dt in enumerate(iter):
    if i >= 10:
        break
    print(dt)

Timezone Handling

import pytz
from datetime import datetime

# Use timezone-aware datetime
tz = pytz.timezone("Europe/Paris")
local_date = tz.localize(datetime(2017, 3, 26))
iter = croniter('0 0 * * *', local_date)

next_match = iter.get_next(datetime)
print(next_match)  # Properly handles DST transitions

Day/Dayofweek Logic

# OR logic (default): run on 1st day of month OR on Wednesdays
iter = croniter('2 4 1 * wed', base, day_or=True)

# AND logic: run on 1st day of month IF it's a Wednesday  
iter = croniter('2 4 1 * wed', base, day_or=False)

Performance Tuning

# Limit search window for sparse expressions
iter = croniter(
    "0 4 1 1 fri",  # 4 AM on January 1st if it's Friday (very sparse)
    datetime(2000, 1, 1),
    day_or=False,
    max_years_between_matches=15  # Limit to 15 years
)

# Use all_next() - won't raise CroniterBadDateError when limit exceeded
for match in iter.all_next(datetime):
    print(match)

Expression Expansion Usage

from croniter import croniter

# Expand a basic cron expression
expanded, field_count = croniter.expand('*/5 0-12 * * mon-fri')
print(f"Fields: {field_count}")
print(f"Expanded: {expanded}")

# Expand with hashed expressions
expanded, _ = croniter.expand('H H * * *', hash_id='unique-job-123')
print(f"Hashed expansion: {expanded}")

# Expand 6-field cron with seconds
expanded, _ = croniter.expand('*/30 */5 0-12 * * mon-fri', second_at_beginning=True)
print(f"6-field expansion: {expanded}")

Static Method Usage

from croniter import croniter

# Check leap years
print(croniter.is_leap(2024))  # True
print(croniter.is_leap(2023))  # False

# Convert month names
numeric_month = croniter.value_alias('jan', croniter.MONTH_FIELD)
print(numeric_month)  # 1

# Convert day names  
numeric_day = croniter.value_alias('mon', croniter.DOW_FIELD)
print(numeric_day)  # 1

Install with Tessl CLI