67 lines
2.0 KiB
Python
67 lines
2.0 KiB
Python
import uuid
|
|
from crontab import CronTab, CronSlices
|
|
|
|
|
|
class CrontabManager:
|
|
COMMENT_PREFIX = "wecker-alarm:"
|
|
|
|
def __init__(self, tabfile: str | None = None, user: bool | str = True):
|
|
# user=True means current user, user="username" means specific user
|
|
self.tabfile = tabfile
|
|
self.user = user
|
|
|
|
def _get_cron(self):
|
|
if self.tabfile:
|
|
return CronTab(tabfile=self.tabfile)
|
|
return CronTab(user=self.user)
|
|
|
|
def get_alarms(self):
|
|
cron = self._get_cron()
|
|
alarms = []
|
|
for job in cron:
|
|
if job.comment.startswith(self.COMMENT_PREFIX):
|
|
alarm_id = job.comment.split(self.COMMENT_PREFIX)[1].strip()
|
|
# job.slices is a valid cron slice object, str(job.slices) gives the expression
|
|
alarms.append(
|
|
{
|
|
"id": alarm_id,
|
|
"cron_expression": str(job.slices),
|
|
"command": job.command,
|
|
"is_enabled": job.is_enabled(),
|
|
}
|
|
)
|
|
return alarms
|
|
|
|
def set_alarm(
|
|
self,
|
|
alarm_id: str | None,
|
|
cron_expression: str,
|
|
command: str,
|
|
is_enabled: bool = True,
|
|
):
|
|
if not CronSlices.is_valid(cron_expression):
|
|
raise ValueError(f"Invalid cron expression: {cron_expression!r}")
|
|
|
|
if not alarm_id:
|
|
alarm_id = str(uuid.uuid4())
|
|
|
|
cron = self._get_cron()
|
|
comment = f"{self.COMMENT_PREFIX}{alarm_id}"
|
|
|
|
# Remove existing if any
|
|
cron.remove_all(comment=comment)
|
|
|
|
# Create new
|
|
job = cron.new(command=command, comment=comment)
|
|
job.setall(cron_expression)
|
|
job.enable(is_enabled)
|
|
|
|
cron.write()
|
|
return alarm_id
|
|
|
|
def delete_alarm(self, alarm_id: str):
|
|
cron = self._get_cron()
|
|
comment = f"{self.COMMENT_PREFIX}{alarm_id}"
|
|
cron.remove_all(comment=comment)
|
|
cron.write()
|