CODEWITHSUNDEEP
celery
dranxo
dranxo

Reputation: 3388

How do I bind a Celery task to a specific queue?

I want to write a task that is only executable from within a given queue - if somebody tries to pass a different queue into the routing_key parameter of apply_async I want to raise an exception. How do I do this?

Upvotes: 3

Views: 3202

Answers (1)

wohlgejm
wohlgejm

Reputation: 140

You could write your own task that would check to make sure a valid routing key is being passed in when apply_async is being called. You can also apply this to queues. Set up routes and queues in your config:

import celery
from kombu import Queue, Exchange

app = celery.Celery('app')
app.conf.CELERY_QUEUES = (
    Queue('add', Exchange('default'), routing_key='good'),
)
app.conf.CELERY_ROUTES = {
    'app.add': {
        'queue': 'add',
        'routing_key': 'good'
    }
}

Now, create your own Task class that will perform the check on the routing key. You'll need to override apply_async:

class RouteCheckerTask(celery.Task):
    abstract = True

    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
                                    link=None, link_error=None, **options):

        app = self._get_app()
        routing_key = options.get('routing_key', None)
        if routing_key:
            valid_routes = [v['routing_key'] for k, v in app.conf.CELERY_ROUTES.items()]
            is_valid = routing_key in valid_routes
            if not is_valid:
                raise NotImplementedError('{} is not a valid routing key. Options are: {}'.format(routing_key, valid_routes))
        if app.conf.CELERY_ALWAYS_EAGER:
            return self.apply(args, kwargs, task_id=task_id or uuid(), link=link, link_error=link_error, **options)
            # add 'self' if this is a "task_method".
        if self.__self__ is not None:
            args = args if isinstance(args, tuple) else tuple(args or ())
            args = (self.__self__, ) + args
        return app.send_task(
            self.name, args, kwargs, task_id=task_id, producer=producer,
            link=link, link_error=link_error, result_cls=self.AsyncResult,
            **dict(self._get_exec_options(), **options)
        )

Base your tasks from this one and call apply_async normally:

@app.task(base=RouteCheckerTask)
def add(x, y):
    return x + y

# Fails
add.apply_async([1, 2], routing_key='bad')
# Passes
add.apply_async([1, 2], routing_key='good')

Upvotes: 1

Related Questions