Support exchange_type argument to apply_async(), defaults to Task.exchange_type

Author: stovenator

celery/execute/__init__.py

@@ -40,6 +40,10 @@ def apply_async(task, args=None, kwargs=None, countdown=None, eta=None,
     :keyword exchange: The named exchange to send the task to. Defaults to
         :attr:`celery.task.base.Task.exchange`.
 
+    :keyword exchange_type: The exchange type to initalize the exchange as
+        if not already declared.
+        Defaults to :attr:`celery.task.base.Task.exchange_type`.
+
     :keyword immediate: Request immediate delivery. Will raise an exception
         if the task cannot be routed to a worker immediately.
         (Do not confuse this parameter with the ``countdown`` and ``eta``
@@ -72,8 +76,10 @@ def apply_async(task, args=None, kwargs=None, countdown=None, eta=None,
     task = tasks[task.name] # get instance from registry
     options = dict(extract_exec_options(task), **options)
     exchange = options.get("exchange")
+    exchange_type = options.get("exchange_type")
 
-    publish = publisher or task.get_publisher(connection, exchange=exchange)
+    publish = publisher or task.get_publisher(connection, exchange=exchange,
+                                              exchange_type=exchange_type)
     try:
         task_id = publish.delay_task(task.name, args, kwargs, task_id=task_id,
                                      countdown=countdown, eta=eta, **options)
@@ -89,7 +95,10 @@ def send_task(name, args=None, kwargs=None, countdown=None, eta=None,
         result_cls=AsyncResult, **options):
 
     exchange = options.get("exchange")
-    publish = publisher or TaskPublisher(connection, exchange=exchange)
+    exchange_type = options.get("exchange_type")
+
+    publish = publisher or TaskPublisher(connection, exchange=exchange,
+                                         exchange_type=exchange_type)
     try:
         task_id = publish.delay_task(name, args, kwargs, task_id=task_id,
                                      countdown=countdown, eta=eta, **options)

celery/task/base.py

@@ -165,6 +165,7 @@ class Task(object):
     rate_limit = conf.DEFAULT_RATE_LIMIT
     rate_limit_queue_type = Queue
     backend = default_backend
+    exchange_type = conf.DEFAULT_EXCHANGE_TYPE
 
     MaxRetriesExceededError = MaxRetriesExceededError
 
@@ -208,7 +209,7 @@ def establish_connection(self,
 
     @classmethod
     def get_publisher(self, connection=None, exchange=None,
-            connect_timeout=conf.BROKER_CONNECTION_TIMEOUT):
+            connect_timeout=conf.BROKER_CONNECTION_TIMEOUT,exchange_type=None):
         """Get a celery task message publisher.
 
         :rtype: :class:`celery.messaging.TaskPublisher`.
@@ -223,9 +224,12 @@ def get_publisher(self, connection=None, exchange=None,
         """
         if exchange is None:
             exchange = self.exchange
+        if exchange_type is None:
+            exchange_type = self.exchange_type
         connection = connection or self.establish_connection(connect_timeout)
         return TaskPublisher(connection=connection,
                              exchange=exchange,
+                             exchange_type=exchange_type,
                              routing_key=self.routing_key)
 
     @classmethod