Async plain Python example

The async equivalent of the plain_python example: a task module, a worker script, and a producer script that schedules work and awaits the result -- built on AsyncAlchemicalQueues / AsyncWorker instead of their sync equivalents.

Run it

cd examples/async_plain_python

In one terminal, start a worker (--with aiosqlite for the async SQLite driver used here):

uv run --with aiosqlite python worker.py "sqlite+aiosqlite:///example.db"

In another terminal, schedule a task and await its result:

uv run --with aiosqlite 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.

There is no alchemical_worker-style CLI for async queues yet, which is why worker.py is a plain script rather than a command -- see its docstring.

tasks.py

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

import asyncio

from alchemical_queues.tasks.aio import async_task
from alchemical_queues.tasks.main import TaskInfo


@async_task
async def add_numbers(taskinfo: TaskInfo, a: int, b: int) -> int:
    print(f"Running task {taskinfo.entry_id}: {a} + {b}")
    # Stand in for an awaitable call (an HTTP request, another query, ...) --
    # this is the kind of work AsyncWorker lets you run concurrently with
    # other tasks on the same event loop, unlike the sync Worker.
    await asyncio.sleep(0)
    return a + b

worker.py

"""Runs an AsyncWorker on the `tasks` queue forever.

There is no `alchemical_worker`-style CLI for async queues yet -- run this
script directly instead, from this directory (so it can `import tasks`):

    python worker.py "sqlite+aiosqlite:///example.db"
"""

import asyncio
import os
import sys
from datetime import timedelta

from sqlalchemy.ext.asyncio import create_async_engine
from tasks import add_numbers  # noqa: F401  (registers the handler by import)

from alchemical_queues.aio import AsyncAlchemicalQueues
from alchemical_queues.tasks.aio import AsyncWorker


async def main() -> None:
    database_url = (
        sys.argv[1]
        if len(sys.argv) > 1
        else os.environ.get("DATABASE_URL", "sqlite+aiosqlite:///example.db")
    )
    engine = create_async_engine(database_url)
    queues = AsyncAlchemicalQueues(engine)
    await queues.create_all()

    task_queue = queues.get_task_queue("tasks")
    worker = AsyncWorker(task_queue, poll_every=timedelta(milliseconds=200))
    await worker.work()


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

producer.py

"""Schedules an `add_numbers` task and awaits its result.

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

    python worker.py "sqlite+aiosqlite:///example.db"

Then run this script:

    python producer.py
"""

import asyncio
import os

from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
from tasks import add_numbers

from alchemical_queues.aio import AsyncAlchemicalQueues


def build_queues(database_url: str) -> AsyncAlchemicalQueues:
    engine: AsyncEngine = create_async_engine(database_url)
    return AsyncAlchemicalQueues(engine)


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

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

    while not await task.done():
        await asyncio.sleep(0.5)

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


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