samuelcolvin
Support for object-oriented `WorkerSettings` classes
In the current implementation, we're only looking at the __dict__ of the user's provided WorkerSettings class, which can be quite misleading if you're trying to get a large multi-queue system set up. The on_startup/on_shutdown hooks are similarly only allowed to be plain functions or @staticmethod, but not @classmethod because those aren't directly callable.
Example of things that could work naturally:
class BaseWorkerSettings:
redis_settings = get_my_common_redis_settings()
@classmethod
async def on_startup(cls, context: dict[str, Any]):
...do some common initialization on all workers...
class FastLane(BaseWorkerSettings):
queue_name = "arq:fast-lane"
tasks = [task_a, task_b]
class SlowLane(BaseWorkerSettings):
queue_name = "arq:slow-lane"
tasks = [task_c, task_d]
@classmethod
async def on_startup(cls, context: dict[str, Any]):
await super().on_startup(context))
...do some specialized initialization on just this worker...
I hit this too. I discovered both the __dict__ and @staticmethod limitations. I've run out of ideas to work around this. My configs are copy/pastes of each other with little bits changed which goes against good principles. Also, I want to be able to subclass a production settings class and change a queue name, so I can write an integration test that won't be affected by a running system on the side.
