Feature Request: Standard way to inject arguments

[AstraLuma]

(sorry for ditching the template)

Things I'm wanting to do with dramatiq involve things shaped like dependency injection: the actor requests a kind of dependency (like a progress status callable or a database connection), and then a middleware accepts the flag and injects the dependency into the message.

This is currently not as easy as it could be, because the message is then re-serialized and dependencies are often not serializable (nor should they be included).

Current solution:

class Injector(dramatiq.Middleware):
    @property
    def actor_options(self):
        return {"give_me_the_thing"}

    def before_process_message(self, broker, message):
        actor = broker.get_actor(message.actor_name)
        if actor.options.get("give_me_the_thing", False):
            message.kwargs['the_thing'] = ...

    def after_process_message(self, broker, message, *, result=None, exception=None):
        try:
            del message.kwargs['the_thing']
        except KeyError:
            pass

    after_skip_message = after_process_message

[hmoravec]

I used this approach for injecting unit of work instance but failed messages were not retried then. Dramatiq emitted standard "Failed to process" message and silently removed it from the queue. However it worked when injecting e.g. integer value. (I added this middleware before Retries middleware.)

[AstraLuma]

Yeah, sounds like Dramatiq couldn't re-serialize the Message.

That's why I delete the additions afterwards.

[Bogdanp]

I think the right level at which to do this is a custom actor decorator that wraps dramatiq's and provides DI. Here's an implementation I had for APIStar:

https://github.com/Bogdanp/apistar_dramatiq/tree/42627a5587639bc3dd7c899328782cbfd95b551b/apistar_dramatiq

I don't think I'd want to standardize something like this in Dramatiq, though, because there are many ways you could want to do DI and Dramatiq already provides enough configuration knobs for you do be able to do it yourself.

[x0day]

Yeah, sounds like Dramatiq couldn't re-serialize the Message.

That's why I delete the additions afterwards.

maybe we can re-serialize message like this #366, use Message.copy method.

re-serialize message is really useful

[AstraLuma]

Ok, I'm trying to wrap my head around how to do this without core dramatiq support, and it's not working.

There's two cases where an actor is called: directly, and from a queue. How I want my application-specific additions to behave will vary by context.

First thing is a status & progress tracker: It reports to a separate data store the status of message, as well as in-job progress (we have some long jobs). It needs the ID of the message to key off in this data store. I want to be able to pass along a callable to the underlying function that'll let it report progress. (If it's from a message, an actual call. If it's a direct call, a filler no-op function.)

Second is database transactions: I want to wrap each call in a transaction, and gives the actor a database handle. In the case of a direct call, it expects the caller to hand them a database handle.

The problem is that I can't come up with a simpleany way to do this without messing around with not-very-extensible parts. It's the buried in the worker code to marshal arguments from the Message to the Actor. Middleware (attached to the broker) is the preferred place to put this, but the natural place to attach additional arguments is to the message.

The core problem is that Message has no ephemeral place, Actor instances are global, and when Actor is actually called, it has no concept of the Message that called it. I can't think of a way to do this without modifying the core worker thread.

[AstraLuma]

Note that I'm not looking for a complete DI framework. I just want a mechanism by which Middleware can add additional keyword arguments to Actors and Messages.