diff --git a/django_q/cluster.py b/django_q/cluster.py
index 824dbaf..8913f50 100644
--- a/django_q/cluster.py
+++ b/django_q/cluster.py
@@ -426,7 +426,6 @@ def save_cached(task, broker):
         iter_count = task.get('iter_count', 0)
         # if it's a group append to the group list
         if group:
-            task_key = '{}:{}:{}'.format(broker.list_key, group, task['id'])
             group_key = '{}:{}:keys'.format(broker.list_key, group)
             group_list = broker.cache.get(group_key) or []
             # if it's an iter group, check if we are ready
diff --git a/django_q/tasks.py b/django_q/tasks.py
index d4e1d66..f933bc3 100644
--- a/django_q/tasks.py
+++ b/django_q/tasks.py
@@ -16,35 +16,35 @@ from django_q.brokers import get_broker
 
 def async(func, *args, **kwargs):
     """Queue a task for the cluster."""
-    # get options from q_options dict or direct from kwargs
-    options = kwargs.pop('q_options', kwargs)
-    broker = options.pop('broker', get_broker())
-    # pop optionals
-    opts = {'hook': None,
-            'group': None,
-            'save': None,
-            'sync': None,
-            'cached': Conf.CACHED,
-            'iter_count': None,
-            'iter_cached': None,
-            'chain': None}
-    for key in opts:
-        opts[key] = options.pop(key, opts[key])
-        # get an id
+    keywords = kwargs.copy()
+    opt_keys = ('hook', 'group', 'save', 'sync', 'cached', 'iter_count', 'iter_cached', 'chain', 'broker')
+    q_options = keywords.pop('q_options', None)
+    # get an id
     tag = uuid()
     # build the task package
-    task = {'id': tag[1], 'name': tag[0],
+    task = {'id': tag[1],
+            'name': tag[0],
             'func': func,
-            'args': args,
-            'kwargs': kwargs,
-            'started': timezone.now()}
+            'args': args}
     # push optionals
-    for key in opts:
-        if opts[key] is not None:
-            task[key] = opts[key]
+    for key in opt_keys:
+        if q_options and key in q_options:
+            task[key] = q_options[key]
+        elif key in keywords:
+            task[key] = keywords.pop(key)
+    # don't serialize the broker
+    broker = task.pop('broker', get_broker())
+    # overrides
+    if 'cached' not in task and Conf.CACHED:
+        task['cached'] = Conf.CACHED
+    if 'sync' not in task and Conf.SYNC:
+        task['sync'] = Conf.SYNC
+    # finalize
+    task['kwargs'] = keywords
+    task['started'] = timezone.now()
     # sign it
     pack = signing.SignedPackage.dumps(task)
-    if task.get('sync', False) or Conf.SYNC:
+    if task.get('sync', False):
         return _sync(pack)
     # push it
     broker.enqueue(pack)
@@ -106,7 +106,7 @@ def result(task_id, wait=0, cached=Conf.CACHED):
         r = Task.get_result(task_id)
         if r:
             return r
-        if (time.time() - start) * 1000 >= wait:
+        if (time.time() - start) * 1000 >= wait >= 0:
             break
         time.sleep(0.01)
 
@@ -122,7 +122,7 @@ def result_cached(task_id, wait=0, broker=None):
         r = broker.cache.get('{}:{}'.format(broker.list_key, task_id))
         if r:
             return signing.SignedPackage.loads(r)['result']
-        if (time.time() - start) * 1000 >= wait:
+        if (time.time() - start) * 1000 >= wait >= 0:
             break
         time.sleep(0.01)
 
@@ -142,14 +142,14 @@ def result_group(group_id, failures=False, wait=0, count=None, cached=Conf.CACHE
     start = time.time()
     if count:
         while True:
-            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0:
                 break
             time.sleep(0.01)
     while True:
         r = Task.get_result_group(group_id, failures)
         if r:
             return r
-        if (time.time() - start) * 1000 >= wait:
+        if (time.time() - start) * 1000 >= wait >= 0:
             break
         time.sleep(0.01)
 
@@ -163,7 +163,7 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non
     start = time.time()
     if count:
         while True:
-            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait > 0:
                 break
             time.sleep(0.01)
     while True:
@@ -175,7 +175,7 @@ def result_group_cached(group_id, failures=False, wait=0, count=None, broker=Non
                 if task['success'] or failures:
                     result_list.append(task['result'])
             return result_list
-        if (time.time() - start) * 1000 >= wait:
+        if (time.time() - start) * 1000 >= wait >= 0:
             break
         time.sleep(0.01)
 
@@ -199,7 +199,7 @@ def fetch(task_id, wait=0, cached=Conf.CACHED):
         t = Task.get_task(task_id)
         if t:
             return t
-        if (time.time() - start) * 1000 >= wait:
+        if (time.time() - start) * 1000 >= wait >= 0:
             break
         time.sleep(0.01)
 
@@ -226,7 +226,7 @@ def fetch_cached(task_id, wait=0, broker=None):
                      result=task['result'],
                      success=task['success'])
             return t
-        if (time.time() - start) * 1000 >= wait:
+        if (time.time() - start) * 1000 >= wait >= 0:
             break
         time.sleep(0.01)
 
@@ -245,14 +245,14 @@ def fetch_group(group_id, failures=True, wait=0, count=None, cached=Conf.CACHED)
     start = time.time()
     if count:
         while True:
-            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+            if count_group(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0:
                 break
             time.sleep(0.01)
     while True:
         r = Task.get_task_group(group_id, failures)
         if r:
             return r
-        if (time.time() - start) * 1000 >= wait:
+        if (time.time() - start) * 1000 >= wait >= 0:
             break
         time.sleep(0.01)
 
@@ -266,7 +266,7 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
     start = time.time()
     if count:
         while True:
-            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait:
+            if count_group_cached(group_id) == count or wait and (time.time() - start) * 1000 >= wait >= 0:
                 break
             time.sleep(0.01)
     while True:
@@ -289,7 +289,7 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None)
                              success=task['success'])
                     task_list.append(t)
             return task_list
-        if (time.time() - start) * 1000 >= wait:
+        if (time.time() - start) * 1000 >= wait >= 0:
             break
         time.sleep(0.01)
 
@@ -472,7 +472,7 @@ class Iter(object):
         if self.started:
             return result(self.id, wait=wait, cached=self.cached)
 
-    def fetch(self,  wait=0):
+    def fetch(self, wait=0):
         """
         get the task result objects.
         :param int wait: how many milliseconds to wait for a result
@@ -560,6 +560,105 @@ class Chain(object):
         return len(self.chain)
 
 
+class Async(object):
+    """
+    an async task
+    """
+
+    def __init__(self, func, *args, **kwargs):
+        self.id = ''
+        self.started = False
+        self.func = func
+        self.args = args
+        self.kwargs = kwargs
+
+    @property
+    def broker(self):
+        return self._get_option('broker', None)
+
+    @broker.setter
+    def broker(self, value):
+        self._set_option('broker', value)
+
+    @property
+    def sync(self):
+        return self._get_option('sync', None)
+
+    @sync.setter
+    def sync(self, value):
+        self._set_option('sync', value)
+
+    @property
+    def save(self):
+        return self._get_option('save', None)
+
+    @save.setter
+    def save(self, value):
+        self._set_option('save', value)
+
+    @property
+    def hook(self):
+        return self._get_option('hook', None)
+
+    @hook.setter
+    def hook(self, value):
+        self._set_option('hook', value)
+
+    @property
+    def group(self):
+        return self._get_option('group', None)
+
+    @group.setter
+    def group(self, value):
+        self._set_option('group', value)
+
+    @property
+    def cached(self):
+        return self._get_option('cached', Conf.CACHED)
+
+    @cached.setter
+    def cached(self, value):
+        self._set_option('cached', value)
+
+    def _set_option(self, key, value):
+        if 'q_options' in self.kwargs:
+            self.kwargs['q_options'][key] = value
+        else:
+            self.kwargs[key] = value
+        self.started = False
+
+    def _get_option(self, key, default=None):
+        if 'q_options' in self.kwargs:
+            return self.kwargs['q_options'].get(key, default)
+        else:
+            return self.kwargs.get(key, default)
+
+    def run(self):
+        self.id = async(self.func, *self.args, **self.kwargs)
+        self.started = True
+        return self.id
+
+    def result(self, wait=0):
+
+        if self.started:
+            return result(self.id, wait=wait, cached=self.cached)
+
+    def fetch(self, wait=0):
+
+        if self.started:
+            return fetch(self.id, wait=wait, cached=self.cached)
+
+    def result_group(self, failures=False, wait=0, count=None):
+
+        if self.started and self.group:
+            return result_group(self.group, failures=failures, wait=wait, count=count, cached=self.cached)
+
+    def fetch_group(self, failures=True, wait=0, count=None):
+
+        if self.started and self.group:
+            return fetch_group(self.group, failures=failures, wait=wait, count=count, cached=self.cached)
+
+
 def _sync(pack):
     """Simulate a package travelling through the cluster."""
     task_queue = Queue()
diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py
index 00bfe19..7fef791 100644
--- a/django_q/tests/test_cached.py
+++ b/django_q/tests/test_cached.py
@@ -5,7 +5,7 @@ import pytest
 from django_q.cluster import pusher, worker, monitor
 from django_q.conf import Conf
 from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached, \
-    async_iter, Chain, async_chain, Iter
+    async_iter, Chain, async_chain, Iter, Async
 from django_q.brokers import get_broker
 
 
@@ -145,3 +145,51 @@ def test_chain(broker):
     # test single
     rid = async_chain(['django_q.tests.tasks.hello', 'django_q.tests.tasks.hello'], sync=True, cached=True)
     assert result_group(rid, cached=True) == ['hello', 'hello']
+
+
+@pytest.mark.django_db
+def test_async_class(broker):
+    broker.purge_queue()
+    broker.cache.clear()
+    a = Async('math.copysign')
+    assert a.func == 'math.copysign'
+    a.args = (1, -1)
+    assert a.started is False
+    a.cached = True
+    assert a.cached is True
+    a.sync = True
+    assert a.sync is True
+    a.broker = broker
+    assert a.broker == broker
+    a.run()
+    assert a.result() == -1
+    assert a.fetch().result == -1
+    # again with kwargs
+    a = Async('math.copysign', 1, -1, cached=True, sync=True, broker=broker)
+    a.run()
+    assert a.result() == -1
+    # with q_options
+    a = Async('math.copysign', 1, -1, q_options={'cached': True, 'sync': False, 'broker': broker})
+    assert a.sync is False
+    a.sync = True
+    assert a.kwargs['q_options']['sync'] is True
+    a.run()
+    assert a.result() == -1
+    a.group = 'async_class_test'
+    assert a.group == 'async_class_test'
+    a.save = False
+    assert a.save is False
+    a.hook = 'djq.tests.tasks.hello'
+    assert a.hook == 'djq.tests.tasks.hello'
+    assert a.started is False
+    a.run()
+    assert a.result_group() == [-1]
+    assert a.fetch_group() == [a.fetch()]
+    # global overrides
+    Conf.SYNC = True
+    Conf.CACHED = True
+    a = Async('math.floor', 1.5)
+    a.run()
+    assert a.result() == 1
+    Conf.SYNC = False
+    Conf.CACHED = False
diff --git a/requirements.txt b/requirements.txt
index 9998ffd..2f76cae 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -6,13 +6,12 @@
 #
 arrow==0.7.0
 blessed==1.12.0
-boto3==1.1.4
-botocore==1.2.11          # via boto3
+boto3==1.2.1
+botocore==1.3.1           # via boto3
 django-picklefield==0.3.2
 django-redis==4.2.0
 docutils==0.12            # via botocore
 future==0.15.2
-futures==2.2.0            # via boto3
 hiredis==0.2.0
 iron-core==1.1.9          # via iron-mq
 iron-mq==0.7