import strawberry from typing import List, Optional import os import sys from pathlib import Path from api.crontab_manager import CrontabManager from common import PID_FILE def get_manager(): tabfile = os.getenv("TABFILE") return CrontabManager(tabfile=tabfile) def is_wecker_ringing() -> bool: """Check if the wecker process is currently running via the PID file.""" if not os.path.exists(PID_FILE): return False try: pid: int with open(PID_FILE) as f: pid = int(f.read().strip()) os.kill(pid, 0) return True except (ValueError, ProcessLookupError, PermissionError, OSError): return False @strawberry.type class Alarm: id: str cron_expression: str command: str is_enabled: bool @strawberry.type class Query: @strawberry.field def is_ringing(self) -> bool: return is_wecker_ringing() @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)