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
• Event Descriptions
• 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


# Each function's docstring is saved as the event's description.
# See "Event Descriptions" below.
@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')

Event Descriptions

Every @event function's docstring is stored alongside the event as its description, so your event store doubles as a catalog of what each event actually does.

@event('every 1 days @ 6:00 am')
def refresh_widget_cache():
    """Refresh the widget cache.

    Pulls the latest widgets and writes them to the cache table.
    """
    ...

The docstring is read with inspect.getdoc() (leading indentation is stripped) and saved to the description column of the cronevents table every time the event is registered:

cronevents=# select func, query, description from cronevents;
         func         |          query           |                      description
----------------------+--------------------------+--------------------------------------------------------
 refresh_widget_cache | every 1 days @ 6:00 am   | Refresh the widget cache.                             +
                      |                          |                                                       +
                      |                          | Pulls the latest widgets and writes them to the cache +
                      |                          | table.

Read it back through the normal API, as CronEvent.description:

from cronevents.settings import get_settings

for cr in get_settings().cronevents.list():
    print(f'{cr.module}.{cr.func} -> {cr.description}')

Notes:

• The description is refreshed on every registration, so editing a docstring updates the stored copy the next time you run cronevents register.
• A function with no docstring stores an empty string, never NULL.
• Event tables created before this feature gain the description column automatically on the next write — there is no manual migration step.
• If you use a custom event store, description is included in CronEvent.to_row(); persist it to get the same behavior.

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):

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, cronevent_id: str | None = None): ...
    def log(self, log: str): ...
    def __exit__(self, exc_type, exc_val, exc_tb): ...
    def list(self, stream: bool = False) -> list[Log]: ...

trigger_id identifies the single run being logged; cronevent_id is the event that run belongs to, so log lines can be found per event as well as per run. The Postgres logger keeps it in an indexed cronevent_id column on cronevents_log. It is optional — a logger whose __init__ only takes trigger_id still works, cronevents passes the second argument only to loggers that accept it.

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

License

• MIT License