Splitting up the Worker and the Queue

[vikigenius]

This is a great library. I love that it uses structured concurrency, I have read great things about it.

The one thing that has caused me some friction/confusion and I think it is apparent in other discussions as well is the worker context.

This appears in multiple places:

It's not very clear that the worker's async context manager doesn't actually start the worker. #94 (reply in thread)

It took me a bit to realize that it does not necessarily mean the worker has started and it's just necessary for enque.

There was also a discussion about modularity of the workers and a potential for task registry.
#44

I think all of these could be solved if the TaskQueue and the Worker were separated. The TaskQueue needs to know how to talk to the broker (Redis). It doesn't care about concurrency or orchestration.

The worker just takes the TaskQueue it wants to process as a parameter and it's generic over the TaskQueueContext, same as the TaskQueue it takes in. It mainly cares about the concurrency, orchestration etc. it doesn't care about the backend TaskQueue implementation

The TaskQueue's async context manager, all it allows you to do is enqueue tasks, you still have to start the worker itself from the CLI.

It's not very different from what you already have, it just more neatly separates concerns and makes for a clearer API imo.

It also might make it more easier to add additional backends.

[Graeme22]

I will look at a PR if you open one. Since v6 hasn't been released yet there's still room for whatever breaking changes to made.

[Graeme22]

I've been thinking a bit about this and I'm not sure it adds much utility. Whichever way you look at it, both the worker and the queue need to have access to the task definitions (unless you don't need types, in which case you can already use enqueue_unsafe and/or type stubs to enqueue jobs without having task definitions). Unless I'm missing something?

[vikigenius]

It's been a while, so I have lost some context. But it will definitely add utility if you want multiple backends.

You’re right that something has to load task definitions for the worker to execute them.

But the Worker API doesn’t need to be coupled to importing task callables at construction time — it only needs the queue’s context type, and the queue determines how tasks are discovered/registered.

TContext = TypeVar("TContext")

class TaskQueue(Generic[TContext]):
    ...

class Worker(Generic[TContext]):
    def __init__(self, queue: TaskQueue[TContext], ...) -> None:
        ...

Then tasks can be defined against a queue instance (or a queue “app” object) and get full context typing:

simple_queue = RedisTaskQueue(context_type=SimpleTaskContext)

@simple_queue.task(timeout=5)
async def fetch(url: str) -> int:
    # simple_queue.context is SimpleTaskContext
    res = await simple_queue.context.http_client.get(url)
    return len(res.text)

And the worker can be constructed purely from the queue it will process:

simple_worker = Worker(queue=simple_queue)

At runtime, task registration can happen by:

importing a module for side effects (e.g., import myapp.tasks.simple), or

an explicit registry/module list passed via CLI/config.

The producer can still opt in for full type safety by importing the task definitions it it wants to enqueue on or use unsafe enqueue, my proposal does not change that

The fully type safe version by importing everything

async with simple_queue:  # The queue will handle connection so you can enqueue, you still need to start worker to process it
    await fetch.enqueue("https://tastyware.dev/")
    # enqueue returns a task object that can be used to get results/info
    task = await fetch.enqueue("https://github.com/tastyware/streaq").start(delay=3)
    print(await task.info())
    print(await task.result(timeout=5))

[Graeme22]

 I'm not particularly interested in other backends, and Redis is pretty tightly integrated into streaQ. Your other points are valid though. For me personally it's not helpful enough to want to write it out but I will seriously consider a PR if you want to contribute.

[Graeme22]

Please see #125, I think this resolves all modularity concerns in a satisfactory way

[vikigenius]

Looks great, we still have to use SAQ because of non idempotent tasks like I mentioned here

#115 but otherwise this PR solves the only other major issue I had with this