import strawberry from typing import List, Optional import os import sys from pathlib import Path from api.crontab_manager import CrontabManager def get_manager(): tabfile = os.getenv("TABFILE") return CrontabManager(tabfile=tabfile) @strawberry.type class Alarm: id: str cron_expression: str command: str is_enabled: bool @strawberry.type class Query: @strawberry.field def get_alarms(self) -> List[Alarm]: manager = get_manager() return [Alarm(**a) for a in manager.get_alarms()] @strawberry.field def get_alarm(self, id: str) -> Optional[Alarm]: manager = get_manager() alarms = manager.get_alarms() for a in alarms: if a["id"] == id: return Alarm(**a) return None @strawberry.type class Mutation: @strawberry.field def set_alarm( self, cron_expression: str, command: Optional[str] = None, is_enabled: bool = True, id: Optional[str] = None, ) -> Alarm: if command is None: project_root = Path(__file__).parent.parent.absolute() python_exec = sys.executable command = f"cd {project_root} && {python_exec} wecker.py >> wecker.log 2>&1" manager = get_manager() new_id = manager.set_alarm( alarm_id=id, cron_expression=cron_expression, command=command, is_enabled=is_enabled, ) return Alarm( id=new_id, cron_expression=cron_expression, command=command, is_enabled=is_enabled, ) @strawberry.field def delete_alarm(self, id: str) -> bool: manager = get_manager() # Verify it exists alarms = manager.get_alarms() exists = any(a["id"] == id for a in alarms) if exists: manager.delete_alarm(id) return True return False schema = strawberry.Schema(query=Query, mutation=Mutation)