feat: declarative cron jobs via pyproject.toml

docs/guides/cron.md

@@ -40,6 +40,29 @@ Expressions use standard five-field cron syntax (`minute hour day-of-month month
 - `queue.delete_cron(task, "*/1 * * * *")` removes a cron; pass the same task and expression you registered.
 - `sheppy cron list` shows registered crons from the CLI.
 
+## Declarative cron jobs (pyproject.toml)
+
+Instead of calling `add_cron()`, you can declare cron jobs in your project's `pyproject.toml`:
+
+```toml
+[[tool.sheppy.cron]]
+task = "myapp.tasks:cleanup"
+expression = "0 3 * * *"
+args = [30]
+
+[[tool.sheppy.cron]]
+task = "myapp.tasks:backup"
+expression = "0 4 * * sun"
+```
+
+Every worker started with `sheppy work` from that directory reads the file and reconciles continuously:
+
+- crons declared in the file are created and marked as file-managed
+- remove an entry from the file and the worker deletes that cron within seconds
+- crons added through `queue.add_cron()` are left alone, even when identical to a declaration
+
+Entries take optional `args`, `kwargs`, and `queue` (which defaults to the worker's first queue). Invalid entries are logged and skipped, and if the file is missing or broken the worker keeps the current state instead of deleting anything. Use `Worker(cron_config_file=...)` for the programmatic path.
+
 ## When to use cron vs `schedule()`
 
 Use `queue.schedule()` for one-off future tasks ("send this email in 30 minutes"). Use `add_cron()` for recurring work ("nightly cleanup at 03:00"). A cron registration persists until you delete it; a scheduled task fires once.

pyproject.toml

@@ -44,6 +44,7 @@ dependencies = [
     "croniter>=6.0.0",
     "pydantic>=2.1.1",
     "redis>=6.0.0,<9.0.0",
+    "tomli>=2.0.0; python_version < '3.11'",
     "typer>=0.19.0",
 ]
 

src/sheppy/_sync_queue.py

@@ -180,6 +180,38 @@ def retry(self, task: Task | UUID | str, at: datetime | timedelta | None = None,
         """
         return self._run_coro(self._queue.retry(task, at, force))
 
+    def cancel(self, task: Task | UUID | str) -> Task:
+        """Cancel a pending or scheduled task.
+
+        Args:
+            task: Instance of a Task or its ID.
+
+        Returns:
+            The updated Task instance with status 'cancelled'.
+
+        Raises:
+            TaskCancellationError: If the task cannot be cancelled - either it
+                was already claimed by a worker, it already finished, or it
+                does not exist.
+        """
+        return self._run_coro(self._queue.cancel(task))
+
+    def delete(self, task: Task | UUID | str) -> bool:
+        """Hard-delete a finished task's stored metadata.
+
+        Only tasks in a terminal state (completed, failed, crashed, cancelled) can be deleted.
+
+        Args:
+            task: Instance of a Task or its ID.
+
+        Returns:
+            True if the task existed and was deleted, False if it was not found.
+
+        Raises:
+            ValueError: If the task has not finished yet.
+        """
+        return self._run_coro(self._queue.delete(task))
+
     def size(self) -> int:
         """Get number of pending tasks in the queue.
 

src/sheppy/_utils/cron_config.py

@@ -0,0 +1,86 @@
+"""
+This file contains utility functions meant for internal use only. Expect breaking changes if you use them directly.
+"""
+
+import logging
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+if sys.version_info >= (3, 11):
+    import tomllib
+else:  # Python 3.10
+    import tomli as tomllib
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class CronDeclaration:
+    task: str
+    expression: str
+    args: tuple[Any, ...] = ()
+    kwargs: dict[str, Any] = field(default_factory=dict)
+    queue: str | None = None
+
+
+def _parse_entry(index: int, entry: Any) -> CronDeclaration | None:
+    if not isinstance(entry, dict):
+        logger.warning(f"[tool.sheppy.cron] entry #{index} is not a table, skipping")
+        return None
+
+    task = entry.get("task")
+    expression = entry.get("expression")
+
+    if not isinstance(task, str) or ":" not in task:
+        logger.warning(f"[tool.sheppy.cron] entry #{index} has a missing or invalid 'task' (expected 'module:function'), skipping")
+        return None
+
+    if not isinstance(expression, str) or not expression.strip():
+        logger.warning(f"[tool.sheppy.cron] entry #{index} ({task}) has a missing or invalid 'expression', skipping")
+        return None
+
+    args = entry.get("args", [])
+    if not isinstance(args, list):
+        logger.warning(f"[tool.sheppy.cron] entry #{index} ({task}) has non-list 'args', skipping")
+        return None
+
+    kwargs = entry.get("kwargs", {})
+    if not isinstance(kwargs, dict):
+        logger.warning(f"[tool.sheppy.cron] entry #{index} ({task}) has non-table 'kwargs', skipping")
+        return None
+
+    queue = entry.get("queue")
+    if queue is not None and not isinstance(queue, str):
+        logger.warning(f"[tool.sheppy.cron] entry #{index} ({task}) has a non-string 'queue', skipping")
+        return None
+
+    return CronDeclaration(task=task, expression=expression, args=tuple(args), kwargs=dict(kwargs), queue=queue)
+
+
+def load_cron_declarations(path: str | Path) -> list[CronDeclaration] | None:
+    path = Path(path)
+
+    try:
+        data = tomllib.loads(path.read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        logger.debug(f"cron config file not found: {path}")
+        return None
+    except (OSError, tomllib.TOMLDecodeError) as e:
+        logger.error(f"cannot parse cron config file {path}: {e}")
+        return None
+
+    entries = data.get("tool", {}).get("sheppy", {}).get("cron", [])
+
+    if not isinstance(entries, list):
+        logger.error(f"[tool.sheppy.cron] in {path} must be an array of tables")
+        return None
+
+    declarations = []
+    for index, entry in enumerate(entries, 1):
+        declaration = _parse_entry(index, entry)
+        if declaration is not None:
+            declarations.append(declaration)
+
+    return declarations

src/sheppy/backend/base.py

@@ -17,7 +17,7 @@ def resolve_metadata_ttl(task_data: dict[str, Any], *, ttl: int | None, error_tt
     task_ttl: TTLValue = config.get("ttl", "inherit")
     task_error_ttl: TTLValue = config.get("error_ttl", "inherit")
 
-    if task_data.get("status") in ("failed", "crashed"):
+    if task_data.get("status") in ("failed", "crashed", "cancelled"):
         for candidate in (task_error_ttl, task_ttl, error_ttl, ttl):
             if not isinstance(candidate, str):  # skip "inherit"
                 return candidate
@@ -77,6 +77,14 @@ async def pop_scheduled(self, queue_name: str, now: datetime | None = None) -> l
     async def store_result(self, queue_name: str, task_data: dict[str, Any]) -> bool:
         pass
 
+    @abstractmethod
+    async def cancel(self, queue_name: str, task_id: str) -> dict[str, Any] | None:
+        pass
+
+    @abstractmethod
+    async def delete_task(self, queue_name: str, task_id: str) -> bool:
+        pass
+
     @abstractmethod
     async def get_results(self, queue_name: str, task_ids: list[str], timeout: float | None = None) -> dict[str, dict[str, Any]]:
         pass

src/sheppy/backend/memory.py

@@ -253,6 +253,51 @@ async def store_result(self, queue_name: str, task_data: dict[str, Any]) -> bool
 
             return True
 
+    async def cancel(self, queue_name: str, task_id: str) -> dict[str, Any] | None:
+        self._check_connected()
+
+        async with self._locks[queue_name]:
+            self._purge_expired(queue_name)
+            task_data = self._task_metadata[queue_name].get(task_id)
+
+            if not task_data or task_data.get("finished_at") is not None:
+                return None
+
+            # task was already claimed by a worker
+            removed = False
+            if task_id in self._pending[queue_name]:
+                self._pending[queue_name].remove(task_id)
+                removed = True
+            else:
+                scheduled = self._scheduled[queue_name]
+                for i, scheduled_task in enumerate(scheduled):
+                    if scheduled_task.task_id == task_id:
+                        scheduled.pop(i)
+                        heapq.heapify(scheduled)
+                        removed = True
+                        break
+
+            if not removed:
+                return None
+
+            task_data["status"] = "cancelled"
+            task_data["finished_at"] = datetime.now(timezone.utc).isoformat()
+            self._set_expiry(queue_name, task_data)
+
+            return deepcopy(task_data)
+
+    async def delete_task(self, queue_name: str, task_id: str) -> bool:
+        self._check_connected()
+
+        async with self._locks[queue_name]:
+            if task_id not in self._task_metadata[queue_name]:
+                return False
+
+            del self._task_metadata[queue_name][task_id]
+            self._task_expiry[queue_name].pop(task_id, None)
+
+            return True
+
     async def get_results(self, queue_name: str, task_ids: list[str], timeout: float | None = None) -> dict[str,dict[str, Any]]:
         self._check_connected()
 

src/sheppy/backend/redis.py

@@ -1,7 +1,7 @@
 import asyncio
 import contextlib
 import json
-from datetime import datetime
+from datetime import datetime, timezone
 from time import time
 from typing import Any, cast
 
@@ -108,6 +108,10 @@ def _pending_tasks_key(self, queue_name: str) -> str:
         """Queued tasks to be processed (stream)"""
         return f"sheppy:pending:{queue_name}"
 
+    def _pending_index_key(self, queue_name: str) -> str:
+        """Index of unclaimed pending tasks: task_id -> stream message_id (hash)"""
+        return f"sheppy:pending_ids:{queue_name}"
+
     def _finished_tasks_key(self, queue_name: str) -> str:
         """Notifications about finished tasks (stream)"""
         return f"sheppy:finished:{queue_name}"
@@ -165,6 +169,7 @@ async def append(self, queue_name: str, tasks: list[dict[str, Any]], unique: boo
         try:
             async with self.client.pipeline(transaction=False) as pipe:
                 pipe.hsetnx(self._queues_registry_key(), queue_name, "{}")
+                xadd_positions = []
 
                 for t in to_queue:
                     _task_data = json.dumps(t)
@@ -173,12 +178,21 @@ async def append(self, queue_name: str, tasks: list[dict[str, Any]], unique: boo
                         pipe.set(f"{tasks_metadata_key}:{t['id']}", _task_data)
 
                     # add to pending stream
+                    xadd_positions.append(len(pipe.command_stack))
                     pipe.xadd(pending_tasks_key, {"data": _task_data})
 
-                await pipe.execute()
+                res = await pipe.execute()
         except Exception as e:
             raise BackendError(f"Failed to enqueue task: {e}") from e
 
+        # record stream message ids in the pending index (used by cancel())
+        if to_queue:
+            mapping = {}
+            for t, pos in zip(to_queue, xadd_positions, strict=True):
+                message_id = res[pos]
+                mapping[t["id"]] = message_id.decode() if isinstance(message_id, bytes) else message_id
+            await self.client.hset(self._pending_index_key(queue_name), mapping=mapping)
+
         return success
 
     async def pop(self, queue_name: str, limit: int = 1, timeout: float | None = None) -> list[dict[str, Any]]:
@@ -212,6 +226,8 @@ async def pop(self, queue_name: str, limit: int = 1, timeout: float | None = Non
                 self._pending_messages[task_data["id"]] = (queue_name, message_id.decode())
                 tasks.append(task_data)
 
+            await self.client.hdel(self._pending_index_key(queue_name), *[t["id"] for t in tasks])
+
             return tasks
 
         except Exception as e:
@@ -246,6 +262,7 @@ async def clear(self, queue_name: str) -> int:
             count += 1
 
         await self.client.xtrim(pending_tasks_key, maxlen=0)
+        await self.client.delete(self._pending_index_key(queue_name))
         await self.client.delete(scheduled_key)
         await self.client.hdel(self._queues_registry_key(), queue_name)
         await self.client.delete(self._rate_limit_key(queue_name))
@@ -354,6 +371,67 @@ async def store_result(self, queue_name: str, task_data: dict[str, Any]) -> bool
         except Exception as e:
             raise BackendError(f"Failed to store task result: {e}") from e
 
+    async def cancel(self, queue_name: str, task_id: str) -> dict[str, Any] | None:
+        tasks_metadata_key = self._tasks_metadata_key(queue_name)
+        scheduled_key = self._scheduled_tasks_key(queue_name)
+        pending_tasks_key = self._pending_tasks_key(queue_name)
+        pending_index_key = self._pending_index_key(queue_name)
+
+        await self._ensure_consumer_group(pending_tasks_key)
+
+        raw = await self.client.get(f"{tasks_metadata_key}:{task_id}")
+        if not raw:
+            return None
+
+        task_data: dict[str, Any] = json.loads(raw)
+
+        if task_data.get("finished_at") is not None:
+            return None
+
+        if task_data.get("status") == "scheduled":
+            removed = await self.client.zrem(scheduled_key, task_id)
+            if removed > 0:
+                return await self._finalize_cancellation(queue_name, task_data)
+            # else: scheduled task just got into pending queue
+
+        raw_message_id = await self.client.hget(pending_index_key, task_id)
+        if raw_message_id is None:
+            # already claimed by a worker (or never queued) -> cancellation fails
+            return None
+
+        message_id = raw_message_id.decode() if isinstance(raw_message_id, bytes) else raw_message_id
+
+        # XDEL is atomic, so only one of two concurrent cancellations wins
+        removed = await self.client.xdel(pending_tasks_key, message_id)
+        await self.client.hdel(pending_index_key, task_id)
+
+        if removed <= 0:
+            # stale index entry -> the message was already removed
+            return None
+
+        return await self._finalize_cancellation(queue_name, task_data)
+
+    async def _finalize_cancellation(self, queue_name: str, task_data: dict[str, Any]) -> dict[str, Any]:
+        task_data["status"] = "cancelled"
+        task_data["finished_at"] = datetime.now(timezone.utc).isoformat()
+
+        async with self.client.pipeline(transaction=True) as pipe:
+            pipe.set(
+                f"{self._tasks_metadata_key(queue_name)}:{task_data['id']}",
+                json.dumps(task_data),
+                ex=resolve_metadata_ttl(task_data, ttl=self.ttl, error_ttl=self.error_ttl),
+            )
+            # notify waiters (wait_for) that the task reached a terminal state
+            min_id = f"{int((time() - self._results_stream_ttl) * 1000)}-0"
+            pipe.xadd(self._finished_tasks_key(queue_name), {"task_id": task_data["id"]}, minid=min_id)
+            await pipe.execute()
+
+        return task_data
+
+    async def delete_task(self, queue_name: str, task_id: str) -> bool:
+        tasks_metadata_key = self._tasks_metadata_key(queue_name)
+        return bool(await self.client.delete(f"{tasks_metadata_key}:{task_id}"))
+
     async def get_stats(self, queue_name: str) -> dict[str, int]:
         scheduled_tasks_key = self._scheduled_tasks_key(queue_name)
         pending_tasks_key = self._pending_tasks_key(queue_name)

src/sheppy/cli/cli.py

@@ -8,6 +8,8 @@
 from .commands.cron.list import list_crons
 from .commands.queue.list import list_queues
 from .commands.task.add import add
+from .commands.task.cancel import cancel
+from .commands.task.delete import delete
 from .commands.task.info import info
 from .commands.task.list import list_tasks
 from .commands.task.retry import retry
@@ -41,6 +43,8 @@ def callback(
 task_app.command()(test)
 task_app.command()(add)
 task_app.command()(schedule)
+task_app.command()(cancel)
+task_app.command()(delete)
 
 app.add_typer(task_app, name="task")