Plain Python example

The smallest possible setup: no web framework at all, just a task module and a script that schedules work and waits for the result. This is the same pattern as the tutorial, kept here as a runnable example.

Run it

In one terminal, start a worker:

cd examples/plain_python
uv run alchemical_worker "sqlite:///example.db" tasks

In another terminal, schedule a task and wait for its result:

cd examples/plain_python
uv run python producer.py

You should see the worker print that it is running the task, and the producer print the result once it is done.

tasks.py

"""The task(s) our worker can run. Needs to be importable by the worker
process under this exact module path, e.g. run `alchemical_worker` from this
directory so it can `import tasks`.
"""

from alchemical_queues.tasks import TaskInfo, task


@task
def add_numbers(taskinfo: TaskInfo, a: int, b: int) -> int:
    print(f"Running task {taskinfo.entry_id}: {a} + {b}")
    return a + b

producer.py

"""Schedules an `add_numbers` task and waits for its result.

Run a worker in another terminal first (from this directory, so it can
import `tasks`):

    alchemical_worker "sqlite:///example.db" tasks

Then run this script:

    python producer.py
"""

import os
import time

from sqlalchemy import create_engine
from tasks import add_numbers

from alchemical_queues import AlchemicalQueues


def build_queues(database_url: str) -> AlchemicalQueues:
    engine = create_engine(database_url)
    queues = AlchemicalQueues(engine)
    queues.create_all()
    return queues


def main() -> None:
    database_url = os.environ.get("DATABASE_URL", "sqlite:///example.db")
    queues = build_queues(database_url)
    task_queue = queues.get_task_queue("tasks")

    task = add_numbers(2, 3).schedule(task_queue)
    print(f"Scheduled task {task.entry_id}, waiting for a worker to pick it up...")

    while not task.done:
        time.sleep(0.5)

    print(f"Result: {task.result}")


if __name__ == "__main__":
    main()