cron.py

This module contains helper stuff to handle adding, editing, and removing cron jobs using one (fairly) simple function.

add(interval)

Add a cron job if it does not already exist, or modify the existing job

Parameters:

Name Type Description Default
interval int

Time (in minutes) between running the program

required

Raises:

Type Description
CNRunError

if we could not run S_CMD_ENV

OSError

If we could not get the values from env

This method first checks to see if the cron job already exists. If it does, we make sure the job is enabled and has the correct interval. If it does not, we add it to the crontab file.

Source code in src/cron.py
def add(interval: int):
    """
    Add a cron job if it does not already exist, or modify the existing job

    Args:
        interval: Time (in minutes) between running the program

    Raises:
        F.CNRunError: if we could not run S_CMD_ENV
        OSError: If we could not get the values from env

    This method first checks to see if the cron job already exists. If it does,
    we make sure the job is enabled and has the correct interval. If it does
    not, we add it to the crontab file.
    """

    # sanitize interval
    interval = int(F.clamp(interval, 1, 59))

    # get current user's crontab
    my_cron = CronTab(user=True)

    # --------------------------------------------------------------------------

    # check if job already exists
    cron_job = _get_job(my_cron)

    # edit if already in there (like swimwear)
    if cron_job:

        # assume nothing changed (no write)
        changed = False

        # if the job is disabled (like if some dumbass edited the crontab file
        # by hand), enable it
        if not cron_job.is_enabled():

            # enable and set change flag
            cron_job.enable(True)
            changed = True

        # if the interval is different (like if some dumbass edited either the
        # config file or the crontab file), set the correct interval
        if _get_interval(cron_job) != interval:

            # set interval and change flag
            cron_job.minute.every(interval)  # type: ignore
            changed = True

        # if either of the previous steps introduced a change, write new
        # crontab file
        if changed:
            my_cron.write()

        # skip the rest of the code
        return

    # --------------------------------------------------------------------------
    # the cron job does not exist, make a new one

    # format cron cmd from values in env
    try:
        cmd = _get_job_cmd()
    except (F.CNRunError, OSError) as e:
        # bubble up the error to be caught by the calling code
        raise e

    # add new job
    my_job = my_cron.new(cmd, S_PRG_NAME)
    my_job.minute.every(interval)  # type: ignore

    # save crontab file
    my_cron.write()

remove()

Remove a cron job if it exists

Source code in src/cron.py
def remove():
    """
    Remove a cron job if it exists
    """

    # get current user's crontab
    my_cron = CronTab(user=True)

    # find our job
    cron_job = _get_job(my_cron)
    if cron_job:

        # remove job
        my_cron.remove(cron_job)

        # save crontab file
        my_cron.write()

update(enable, interval)

Update an existing job if the config file changes

Parameters:

Name Type Description Default
enable bool

Whether to enable or disable the job

required
interval int

New interval for the job (default: None)

required

This method updates the crontab file based on the given parameters. If you want to add/update a cron job, pass "enable=True" and "interval=". If interval is None, the current (or default) value will be used.

Source code in src/cron.py
def update(enable: bool, interval: int):
    """
    Update an existing job if the config file changes

    Args:
        enable: Whether to enable or disable the job
        interval: New interval for the job (default: None)

    Raises:
        OSError if enable is True and interval is None

    This method updates the crontab file based on the given parameters. If you
    want to add/update a cron job, pass "enable=True" and "interval=<int>". If
    interval is None, the current (or default) value will be used.
    """

    # sanitize enable
    # TODO: bool cast gone in next cnlib update
    enable = bool(F.bool(enable))

    # sanitize interval
    if interval:
        interval = int(F.clamp(interval, 1, 59))

    # cfg wants us to enable
    if enable:

        # must have interval when enabling
        if interval:
            add(interval)
        else:
            print("Hey CN: if enabling, interval must be specified")

    # cfg wants disable
    else:
        remove()