Sheppy¶

Sheppy is an async-native task queue for Python. Decorate a function with @task, add it to a Queue, and a worker process executes it. Arguments are validated with Pydantic, tasks are stored in Redis (and soon Postgres), and workers wait on blocking reads instead of polling.

import asyncio
from sheppy import Queue, task

@task
async def send_email(to: str, subject: str) -> str:
    return f"sent '{subject}' to {to}"

async def main() -> None:
    queue = Queue("memory://")  # use "redis://localhost:6379" in production
    t = send_email("[email protected]", "Welcome!")  # returns a Task; nothing runs yet
    await queue.add(t)
    done = await queue.wait_for(t)
    print(done.result)

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

$ python example.py
sent 'Welcome!' to [email protected]

The memory:// backend executes tasks in-process, so this example needs no external services. In production you point the queue at Redis and run one or more worker processes with sheppy work.

Where to go next¶

• Quickstart: install, first task, first worker, in about five minutes.
• How-to guides: scheduling, cron, retries, rate limiting, workflows, dependency injection, testing.
• API reference: exact signatures and behavior of Queue, Worker, TestQueue, and the models.
• Code examples: runnable scripts from the repository, each with the command to run it.