refactor(cron): extract _format_timing and _format_state helpers

Addresses review feedback: moves schedule formatting and state
formatting into dedicated static methods, removes duplicate
in-loop imports, and simplifies _list_jobs() to a clean loop.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
PJ Hoberman
2026-03-17 15:03:30 +00:00
committed by Xubin Ren
parent 228e1bb3de
commit 8d45fedce7

View File

@@ -1,11 +1,12 @@
"""Cron tool for scheduling reminders and tasks."""
from contextvars import ContextVar
from datetime import datetime, timezone
from typing import Any
from nanobot.agent.tools.base import Tool
from nanobot.cron.service import CronService
from nanobot.cron.types import CronSchedule
from nanobot.cron.types import CronJobState, CronSchedule
class CronTool(Tool):
@@ -143,49 +144,49 @@ class CronTool(Tool):
)
return f"Created job '{job.name}' (id: {job.id})"
@staticmethod
def _format_timing(schedule: CronSchedule) -> str:
"""Format schedule as a human-readable timing string."""
if schedule.kind == "cron":
tz = f" ({schedule.tz})" if schedule.tz else ""
return f"cron: {schedule.expr}{tz}"
if schedule.kind == "every" and schedule.every_ms:
secs = schedule.every_ms // 1000
if secs >= 3600:
return f"every {secs // 3600}h"
if secs >= 60:
return f"every {secs // 60}m"
return f"every {secs}s"
if schedule.kind == "at" and schedule.at_ms:
dt = datetime.fromtimestamp(schedule.at_ms / 1000, tz=timezone.utc)
return f"at {dt.isoformat()}"
return schedule.kind
@staticmethod
def _format_state(state: CronJobState) -> list[str]:
"""Format job run state as display lines."""
lines: list[str] = []
if state.last_run_at_ms:
last_dt = datetime.fromtimestamp(state.last_run_at_ms / 1000, tz=timezone.utc)
info = f" Last run: {last_dt.isoformat()}{state.last_status or 'unknown'}"
if state.last_error:
info += f" ({state.last_error})"
lines.append(info)
if state.next_run_at_ms:
next_dt = datetime.fromtimestamp(state.next_run_at_ms / 1000, tz=timezone.utc)
lines.append(f" Next run: {next_dt.isoformat()}")
return lines
def _list_jobs(self) -> str:
jobs = self._cron.list_jobs()
if not jobs:
return "No scheduled jobs."
lines = []
for j in jobs:
s = j.schedule
if s.kind == "cron":
timing = f"cron: {s.expr}"
if s.tz:
timing += f" ({s.tz})"
elif s.kind == "every" and s.every_ms:
secs = s.every_ms // 1000
if secs >= 3600:
timing = f"every {secs // 3600}h"
elif secs >= 60:
timing = f"every {secs // 60}m"
else:
timing = f"every {secs}s"
elif s.kind == "at" and s.at_ms:
from datetime import datetime, timezone
dt = datetime.fromtimestamp(s.at_ms / 1000, tz=timezone.utc)
timing = f"at {dt.isoformat()}"
else:
timing = s.kind
timing = self._format_timing(j.schedule)
status = "enabled" if j.enabled else "disabled"
parts = [f"- {j.name} (id: {j.id}, {timing}, {status})"]
if j.state.last_run_at_ms:
from datetime import datetime, timezone
last_dt = datetime.fromtimestamp(j.state.last_run_at_ms / 1000, tz=timezone.utc)
last_info = (
f" Last run: {last_dt.isoformat()}{j.state.last_status or 'unknown'}"
)
if j.state.last_error:
last_info += f" ({j.state.last_error})"
parts.append(last_info)
if j.state.next_run_at_ms:
from datetime import datetime, timezone
next_dt = datetime.fromtimestamp(j.state.next_run_at_ms / 1000, tz=timezone.utc)
parts.append(f" Next run: {next_dt.isoformat()}")
parts.extend(self._format_state(j.state))
lines.append("\n".join(parts))
return "Scheduled jobs:\n" + "\n".join(lines)