Task scheduling¶

queue.schedule() stores a task until its time comes; the scheduler inside every running worker then moves it into the pending queue for execution.

Schedule with a delay¶

Pass a timedelta to run a task relative to now:

from datetime import timedelta

await queue.schedule(send_email(email), at=timedelta(minutes=30))

Schedule at an absolute time¶

Pass a timezone-aware datetime:

from datetime import datetime, timezone

await queue.schedule(send_email(email), at=datetime(2027, 1, 1, tzinfo=timezone.utc))

Note

Datetime must have timezone included. A naive datetime (one without tzinfo) raises TypeError.

What happens next¶

The task is stored with status scheduled. queue.get_scheduled() lists all scheduled tasks.
Every running worker checks for due tasks once per second and moves them into the pending queue. Expect execution up to about a second after the scheduled time. This is not a real-time scheduler.
If no worker is running, the task waits in Redis until one starts.

Full example¶

examples/simple_scheduled.py schedules a follow-up email two seconds after a welcome email, using Redis backend (don't forget to start the worker with sheppy work in a separate terminal):

examples/simple_scheduled.py

import asyncio
from datetime import datetime, timedelta

from pydantic import BaseModel

from sheppy import Queue, task


class Email(BaseModel):
    to: str
    subject: str
    body: str


@task
async def send_email(email: Email) -> dict[str, str]:
    print(f"[{datetime.now()}] Sent email to {email.to},"
          f"Subject: {email.subject}, Body: {email.body}")
    return {"status": "sent"}


queue = Queue("redis://localhost:6379")


async def main():
    welcome_email = Email(to="[email protected]",
                          subject="Registration Successful!",
                          body="Your account has been created!")

    survey_email = Email(to="[email protected]",
                         subject="Feedback survey",
                         body="How do you like our new website?")


    # send welcome email immediately
    task = send_email(welcome_email)
    await queue.add(task)

    # schedule feedback email to be delivered after
    # a few weeks (2 seconds in this example)
    survey_email_task = send_email(survey_email)
    await queue.schedule(survey_email_task, at=timedelta(seconds=2))

    # wait for results to verify welcome email was sent
    task = await queue.wait_for(task)
    assert task.result.get("status", None) == "sent"
    assert task.status == 'completed'
    assert not task.exception

    # confirm scheduled task wasn't sent yet
    survey_email_task = await queue.get_task(survey_email_task)
    print(survey_email_task.status)
    assert survey_email_task.status == 'scheduled'

    # wait for scheduled email to happen (for demo purposes)
    print("sleeping for 3 seconds...")
    await asyncio.sleep(3)

    # verify the email was sent (for demo purposes)
    survey_email_task = await queue.get_task(survey_email_task)
    assert survey_email_task.status == 'completed'
    assert not survey_email_task.exception
    assert survey_email_task.result.get("status") == "sent"


if __name__ == "__main__":
    asyncio.run(main())

Output should look something like this:

$ python examples/simple_scheduled.py
[2026-07-19 14:56:01.127391] Sent email to [email protected], Subject: Registration Successful!, Body: Your account has been created!
sleeping for 3 seconds...
[2026-07-19 14:56:03.139961] Sent email to [email protected], Subject: Feedback survey, Body: How do you like our new website?

Schedule from the CLI¶

sheppy task schedule myapp.tasks:send_email --delay 30m --kwargs '{"to": "[email protected]"}'
sheppy task schedule myapp.tasks:send_email --at 2027-01-01T09:00:00

--delay accepts values like 30s, 5m, 2h, or 1d; --at takes an ISO 8601 datetime, and naive values are treated as UTC. The two options are mutually exclusive.

Retry a failed task at a later time with queue.retry(task, at=timedelta(hours=1)) (see Error handling).
Recurring tasks: Cron jobs.