cronevents 0.0.45

A package to run cron jobs(events)

Cron Events

This module provides a way to schedule recurring events using a cron-like syntax. Events can be scheduled to run at specific intervals or at specific times on specific days.

Table of Contents

• Installation
• Learn by Example
• Configuration
• License

Installation

pip install cronevents

Learn by Example

"""Cron-like event scheduling module.

This module provides a way to schedule recurring events using a cron-like syntax.
Events can be scheduled to run at specific intervals or at specific times on specific days.

Syntax:
    '(`every` | `in` | `on`) (`Weekday` | `n unit [n unit ...]`) [@ hh[:mm[:ss]] ["am" | "pm"]] [Timezone]'

    Combine multiple schedules with `||`:
    'every Monday @ 9:00 am || every Friday @ 5:00 pm'

Examples:
    'every 2 days @ 10:00:00 pm'
    'every Monday @ 23'
    'every 5 seconds'
    'every 2 days 1 hours 23 minutes 2 seconds'
    'every 1 days @ 9:00 am America/New_York'

Note:
    Using '@' will run the event at least once a day.
    Timezone defaults to UTC. Append any IANA timezone name (e.g. America/New_York)
    or a custom abbreviation registered with add_timezone_abbr().
"""

# Uncomment to register events to the event manager
# import os
# os.environ['REGISTER_CRON_EVENT'] = 'true'

from cronevents.event_manager import event, add_timezone_abbr


@event('every 31 seconds')
def test():
    """Write 'test' to a file and print it every 31 seconds."""
    with open('test.txt', 'a') as f:
        f.write('test\n')
    print('test')


@event('every 2 days 1 hours 23 minutes 2 seconds')
def test2():
    """Write 'test2' to a file and print 'test2' every 2 days, 1 hour, 23 minutes, and 2 seconds."""
    with open('test.txt', 'a') as f:
        f.write('test2\n')
    print('test2')


@event('every 1 days @ 2:00 pm')
def test3():
    """Write 'test3' to a file and print it daily at 2:00 PM."""
    with open('test.txt', 'a') as f:
        f.write('test3\n')
    print('test3')


@event('every Friday')
def test4():
    """Write 'test4' to a file and print it every Friday."""
    with open('test.txt', 'a') as f:
        f.write('test4\n')
    print('test4')


@event('every Tuesday @ 3:00')
def test5():
    """Write 'test5' to a file and print it every Tuesday at 3:00 AM."""
    with open('test.txt', 'a') as f:
        f.write('test5\n')
    print('test5')


# Use a full IANA timezone name appended to the query
@event('every 1 days @ 9:00 am America/New_York')
def test6():
    """Write 'test6' to a file and print it daily at 9:00 AM Eastern time."""
    with open('test.txt', 'a') as f:
        f.write('test6\n')
    print('test6')


# Or register a custom abbreviation and use that instead
add_timezone_abbr('America/Los_Angeles', 'PT')

@event('every Monday @ 8:00 am PT')
def test7():
    """Write 'test7' to a file and print it every Monday at 8:00 AM Pacific time."""
    with open('test.txt', 'a') as f:
        f.write('test7\n')
    print('test7')

Configuration

On first run, cronevents auto-creates a settings file at .cronevents/settings.yaml with the defaults shown below. You can also create it by running cronevents init. Edit this file to change backends or toggle logging. The file location can be overridden with the CRONEVENTS_SETTINGS_PATH environment variable.

Default settings.yaml

log_cronevents_triggers: true   # record every time an event fires
log_cronevents_processes: false # capture stdout/stderr of event subprocesses

cronevents:                     # where registered events are stored
  module: cronevents.db.cronevents.sqlite
  name: Sqlite3CronEventsDb

logger:                         # how subprocess output is stored (when log_cronevents_processes: true)
  module: cronevents.db.logs.file
  name: FileLogger

trigger:                        # where trigger history is stored (when log_cronevents_triggers: true)
  module: cronevents.db.triggers.sqlite
  name: Sqlite3TriggerDb

How It Works

When anything imports from cronevents, get_settings() is called lazily on first use. It reads settings.yaml (or creates it if missing), then constructs a Settings object. Each of the three backend keys (cronevents, logger, trigger) is a module/name pair pointing to a class that cronevents will import and instantiate at runtime. If the class can't be loaded for any reason, it falls back to the SQLite/file defaults.

Using PostgreSQL Backends

Switch any or all backends to Postgres by updating settings.yaml:

log_cronevents_triggers: true
log_cronevents_processes: true

cronevents:
  module: cronevents.db.cronevents.postgres
  name: PostgresCronEventsDb

logger:
  module: cronevents.db.logs.postgres
  name: PostgresLogger

trigger:
  module: cronevents.db.triggers.postgres
  name: PostgresTriggerDb

The Postgres backends read connection details from environment variables (or a .env file):

Variable	Default
POSTGRES_HOST	localhost
POSTGRES_PORT	5432
POSTGRES_USER	—
POSTGRES_PASSWORD	—
POSTGRES_DATABASE	—

Custom Backends

You can point any backend at your own class — just subclass the appropriate base and reference it in settings.yaml.

Custom event store — subclass CronEventsDbBase:

# myproject/my_store.py
from cronevents.db.cronevents.base import CronEventsDbBase, CronEvent

class MyCronEventsDb(CronEventsDbBase):
    def insert(self, cronevent: CronEvent) -> None: ...
    def update(self, cronevent: CronEvent) -> None: ...
    def upsert(self, cronevent: CronEvent) -> None: ...
    def delete(self, cronevent_id: str) -> None: ...
    def get(self, cronevent_id=None, module=None, func=None) -> CronEvent | None: ...
    def list(self) -> list[CronEvent]: ...

Custom logger — subclass LoggerBase:

# myproject/my_logger.py
from cronevents.db.logs.base import LoggerBase, Log

class MyLogger(LoggerBase):
    def __init__(self, trigger_id: str): ...
    def log(self, log: str): ...
    def __exit__(self, exc_type, exc_val, exc_tb): ...
    def list(self, stream: bool = False) -> list[Log]: ...

Custom trigger store — subclass TriggerDbBase:

# myproject/my_triggers.py
from cronevents.db.triggers.base import TriggerDbBase, Trigger

class MyTriggerDb(TriggerDbBase):
    def insert(self, trigger: Trigger): ...
    def upsert(self, trigger: Trigger): ...
    def list(self, stream=False, cronevent_id=None) -> list[Trigger]: ...

Then reference your class in settings.yaml:

cronevents:
  module: myproject.my_store
  name: MyCronEventsDb

logger:
  module: myproject.my_logger
  name: MyLogger

trigger:
  module: myproject.my_triggers
  name: MyTriggerDb

Environment Variables

Variable	Default	Description
CRONEVENTS_SETTINGS_PATH	.cronevents/settings.yaml	Path to the settings file
CRONEVENTS_LOG_DIR	.cronevents/logs	Directory for file-based process logs
REGISTER_CRON_EVENT	false	Set to true to register events on import
POSTGRES_HOST	localhost	Postgres host (Postgres backends only)
POSTGRES_PORT	5432	Postgres port
POSTGRES_USER	—	Postgres user
POSTGRES_PASSWORD	—	Postgres password
POSTGRES_DATABASE	—	Postgres database name

License

• MIT License