[Feature] Add Task Registry

[funkybob]

Currently a WIP, this adds a Registry class to allow registering tasks independently of a Broker, and later register them with a given broker.

This solves the chicken-and-egg problem where decorated task functions may be imported before the configuration for the Broker has been loaded and applied.

Still needs tests...

[aberres]

Suggested change

	This allows your code to declar Actors before configuring a Broker.
	This allows your code to declare actors before configuring a broker.

[aberres]

Great to see this! I nearly have the same class in my projects.

I have implemented this method as well:

    def enqueue(self, message, *, delay=None):
        raise RuntimeError(
            "ActorCollector has not transferred its actors to an actual broker. "
            "Ensure transfer_actors() is called."
        )

It ensures that the user is told what is going on when trying to call send() on an actor before doing the bind.

[aberres]

In my code I have an additional helper method to allow binding multople collectors (here it would be registries) at once.

This is handy when linking up independet code from multiple packages.

def transfer_actors(broker, collectors: List[ActorCollector]):
    for ac in collectors:
        ac.transfer_actors(broker)

[ashleybartlett]

I have been able to achieve something like this by defining a middleware, that effectively looks like:

APIActors = Dict[str, Actor
API_ACTORS: APIActors = {}

class APIActionMiddleware(Middleware):
    """Middleware that exposes tasks to be executed by the API

        >>> @actor(api=True)
        ... def example():
        ...     do_stuff()

    The task can then be used in the API by whatever name you have set for this task. In the above
    example, it would be `example`. If you override the task name using `actor_name` it will use
    that. In this example, the task will be registered under `do-stuff`.

        >>> @actor(actor_name="do-stuff", api=True)
        ... def example():
        ...     do_stuff()

    Parameters:
        api(bool): Whether to allow this task to be exposed to the API. Pass this in as an extra
            option when using the `@actor(api=True)` decorator.
            Defaults to `False`.
    """

    @property
    def actor_options(self):
        return {"api"}

    def after_declare_actor(self, broker: Broker, actor: Actor) -> None:
        # pop this, otherwise it gets sent with every message which isn't necessary.
        if actor.options.pop("api", False):
            LOGGER.debug("Actor available to API %r.", actor.actor_name)
            API_ACTORS[actor.actor_name] = actor

Then I can reference API_ACTORS from anywhere in my code. Not sure if this approach works. I wanted to optionally store tasks for running via an API.

[ashleybartlett]

Ah, but I have missed the point about doing this separately from a broker. Never mind.

[AstraLuma]

flask-melodramatiq solves this as well.