Quickstart¶

Install Sheppy, define your first task, and watch a worker process it.

Installation¶

pip install sheppy
# or if you're using uv:
uv add sheppy
source .venv/bin/activate

Your First Task¶

You will build a task that adds two numbers: enough to see the full flow from enqueue to result.

Step 0: Import Required Modules¶

Create a file called quickstart.py with these imports:

quickstart.py

import asyncio
from sheppy import Queue, task

Step 1: Define a Task¶

Add the @task decorator to your function:

quickstart.py

@task
async def calculate(x: int, y: int) -> int:
    return x + y

The decorator changes what calling the function means: calculate(2, 1) no longer executes the function. It returns a Task instance that can be queued.

Tip

Tasks can be sync or async. Sheppy handles both. See Sync vs Async Tasks below.

Step 2: Create a Queue¶

Queues need a backend for task storage. Use Redis:

quickstart.py

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

Tip

Start Redis with Docker: docker run -d --name redis -p 6379:6379 redis:7

Note

Sheppy ships with Redis and in-memory backends. More backends are on the Roadmap, or implement your own by extending the Backend class.

Step 3: Add Tasks to the Queue¶

Sheppy is async-first, so wrap your queue operations in an async function:

async def main():
    # create task instances (returns Task object)
    t = calculate(1, 2)

    # add task to the queue
    await queue.add(t)
    print(f"Task {t.id} added to the queue.")

Calling calculate(1, 2) creates a Task instance. Adding it to the queue makes it available for workers to process.

Step 4: Wait for Task Completion¶

Use wait_for() to block until a worker processes the task:

async def main():
    # ... previous code ...

    # wait for task to be processed and get the result
    processed = await queue.wait_for(t)

    if processed.status == 'completed':
        print(f"Task {t.id} completed with result: {processed.result}")
    elif processed.exception:
        print(f"Task {t.id} failed with error: {processed.exception}")
    else:
        # this shouldn't happen because we are waiting for the task to complete
        print(f"Task {t.id} is still pending.")

The timeout parameter controls how long to wait. In a web application you would usually return the task id to the client immediately and expose status through a separate endpoint instead of blocking.

Step 5: Run the Script¶

Add the entry point to run your async main function:

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

Run it:

python quickstart.py

The script hangs waiting for a worker. Start one next.

Step 6: Start a Worker¶

In a separate terminal, start a worker:

sheppy work --backend-url redis://localhost:6379

The worker picks up and processes the task:

Second terminal output

bash:~$ sheppy work --backend-url redis://localhost:6379
Starting worker for queue 'default'
  Backend: RedisBackend [redis://localhost:6379]
  Job processing: True  Scheduler: True  Cron Manager: True
  Max concurrent tasks: 10

2026-01-01T15:01:45Z INFO     scheduler Started
2026-01-01T15:01:45Z INFO     cron      Started
2026-01-01T15:01:48Z INFO     worker    Task started function=quickstart:calculate queue=default task_id=074396c1-e11f-40a3-b22b-094dc89573ea
2026-01-01T15:01:48Z INFO     worker    Task completed function=quickstart:calculate queue=default task_id=074396c1-e11f-40a3-b22b-094dc89573ea

Back in the first terminal, you see the result:

First terminal output

bash:~$ python quickstart.py
Task 074396c1-e11f-40a3-b22b-094dc89573ea added to the queue.
Task 074396c1-e11f-40a3-b22b-094dc89573ea completed with result: 3
bash:~$

(The task id and timestamps differ on every run.)

Complete Example¶

Here is the full quickstart.py for reference:

quickstart.py

import asyncio
from sheppy import Queue, task


@task
async def calculate(x: int, y: int) -> int:
    return x + y


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


async def main():
    # create task instances (returns Task object)
    t = calculate(1, 2)

    # add task to the queue
    await queue.add(t)
    print(f"Task {t.id} added to the queue.")

    # wait for task to be processed and get the result
    processed = await queue.wait_for(t)

    if processed.status == 'completed':
        print(f"Task {t.id} completed with result: {processed.result}")
    elif processed.exception:
        print(f"Task {t.id} failed with error: {processed.exception}")
    else:
        # this shouldn't happen because we are waiting for the task to complete
        print(f"Task {t.id} is still pending.")


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

Sync vs Async Tasks¶

Sheppy handles both async and sync tasks:

@task
async def async_task(x: int) -> int:
    await asyncio.sleep(1)
    return x * 2

@task
def sync_task(x: int) -> int:
    time.sleep(1)  # blocking operation
    return x * 2

Sync tasks run in a thread pool so they do not block the worker's event loop.