Repository: python-telegram-bot/python-telegram-bot
Pull request: Refactor @run_async
* Deprecates the @run_async decorator
* Instead adds kwargs `update` and `error_handler` to `DP.run_async()` to allow proper error handling
* Allows to specify that a callback should be run asynchronously via `DP.add_handler`/`DP.add_error_handler`

One problem: In the new setting it's currently not possible to make the handlers of a `ConversationHandler` be run asynchronously. Possible solutions:
* When adding a CH via `DP.add_handler(CH, run_asyn=True)` make *all* handlers of the Conversation async
* Set the `run_async` flag as parameter for `Handler` instead of as parameter for `DP.add_handler`
* Somehow build an option for that into CH

Wanted to have that clarified before going for the tests, so WIP.

When ready, closes #682 and superseeds/closes #785

Problem statement:
Error handlers not working with @run_async
### Steps to reproduce
1. Register any handler, for ex: `CommandHandler`;
2. Write some code to get `TelegramException` (ex: invalid Markdown/HTML);
3. Register error handler;
4. Run it.

Example code:

```python
import logging

from telegram import Update, Bot, ParseMode
from telegram.ext import Updater, CommandHandler
from telegram.ext.dispatcher import run_async

updater = Updater('<telegram_token>')

logging.basicConfig(
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    level=logging.INFO
)
logger = logging.getLogger(__name__)

dispatcher = updater.dispatcher

msg = "<invalid_tag>Bold line"


def action_start(bot: Bot, update: Update):
    logger.info('/start')
    update.message.reply_text(msg, parse_mode=ParseMode.HTML)


@run_async
def action_async_start(bot: Bot, update: Update):
    logger.info('/async_start')
    update.message.reply_text(msg, parse_mode=ParseMode.HTML)


def _error(bot: Bot, update: Update, e: BaseException):
    logger.error("Boom!")


dp = updater.dispatcher

dp.add_handler(CommandHandler('start', action_start))
dp.add_handler(CommandHandler('async_start', action_async_start))

dp.add_error_handler(_error)

updater.start_polling()
updater.idle()

```


### Expected behavior
Error handler will catch `TelegramException`, so handler could manage it.

### Actual behavior
Error handler is not called.

### Configuration
**Version of Python, python-telegram-bot & dependencies:**
```
python-telegram-bot 6.0.2
urllib3 1.21.1
certifi 2017.04.17
future 0.16.0
Python 3.5.2 (v3.5.2:4def2a2901a5, Jun 26 2016, 10:47:25)  [GCC 4.2.1 (Apple Inc. build 5666) (dot 3)]
```

### Logs
Commands to send:
```
/start
/async_start
```

Output:
```
<project_path>/venv/bin/python <project_path>/issue.py
2017-06-20 23:24:11,743 - __main__ - INFO - /start
2017-06-20 23:24:11,906 - telegram.ext.dispatcher - WARNING - A TelegramError was raised while processing the Update.
2017-06-20 23:24:11,907 - __main__ - ERROR - Boom!
2017-06-20 23:24:11,907 - __main__ - INFO - /async_start
2017-06-20 23:24:12,188 - telegram.utils.promise - ERROR - An uncaught error was raised while running the promise
Traceback (most recent call last):
  File "<project_path>/venv/src/python-telegram-bot/telegram/utils/promise.py", line 42, in run
    self._result = self.pooled_function(*self.args, **self.kwargs)
  File "<project_path>/issue.py", line 25, in action_async_start
    update.message.reply_text(msg, parse_mode=ParseMode.HTML)
  File "<project_path>/venv/src/python-telegram-bot/telegram/message.py", line 340, in reply_text
    return self.bot.send_message(self.chat_id, *args, **kwargs)
  File "<project_path>/venv/src/python-telegram-bot/telegram/bot.py", line 125, in decorator
    result = func(self, *args, **kwargs)
  File "<project_path>/venv/src/python-telegram-bot/telegram/bot.py", line 158, in decorator
    return Bot._message_wrapper(self, url, data, *args, **kwargs)
  File "<project_path>/venv/src/python-telegram-bot/telegram/bot.py", line 146, in _message_wrapper
    result = self._request.post(url, data, timeout=kwargs.get('timeout'))
  File "<project_path>/venv/src/python-telegram-bot/telegram/utils/request.py", line 252, in post
    **urlopen_kwargs)
  File "<project_path>/venv/src/python-telegram-bot/telegram/utils/request.py", line 194, in _request_wrapper
    raise BadRequest(message)
telegram.error.BadRequest: Can't parse entities in message text: unsupported start tag "invalid_tag" at byte offset 0

Process finished with exit code 0
```

I suppose, `dispatcher.dispatch_error` should be called in the same thread with async handler.

Hints from issue discussion:
Hi there, 

Thanks for the report. We'll look into this.
same problem, still not fix?
Guys! I still have a problem! After I added
``` python
from telegram.utils import request
if request.is_con_pool_initialized():
    raise RuntimeError('this is not prior to anything else...')
request.CON_POOL_SIZE = 20
```

I got the error: 

``` python
Traceback (most recent call last):
  File "bot.py", line 1, in <module>
    from frontend.tgbot.main_functions import main
  File "------------------------/frontend/tgbot/main_functions.py", line 35, in <module>
    if request.is_con_pool_initialized():
AttributeError: module 'telegram.utils.request' has no attribute 'is_con_pool_initialized'
```

What should I do now? 
@ohld, you are probably using a quite old guide (or docs) for python-telegram-bot, this function was [removed](https://github.com/python-telegram-bot/python-telegram-bot/commit/e4a132c0e41b7494c0c68da39895f6f6305e74c7#diff-645ee0d992c7b0c7add038d83c52875f) for more then 2 years now
aaand, probably, miss the topic

Commit message: Revert "Remove remnant of previous change"

This reverts commit e4b5e2e2

Patch to review:
diff --git a/telegram/ext/callbackcontext.py b/telegram/ext/callbackcontext.py
index 5dc8aac3032..b4800fb2125 100644
--- a/telegram/ext/callbackcontext.py
+++ b/telegram/ext/callbackcontext.py
@@ -38,7 +38,8 @@ class CallbackContext:
         use a fairly unique name for the attributes.
 
     Warning:
-         Do not combine custom attributes and @run_async. Due to how @run_async works, it will
+         Do not combine custom attributes and ``@run_async``/
+         :meth:`telegram.ext.Disptacher.run_async`. Due to how @run_async works, it will
          almost certainly execute the callbacks for an update out of order, and the attributes
          that you think you added will not be present.
 
diff --git a/telegram/ext/callbackqueryhandler.py b/telegram/ext/callbackqueryhandler.py
index 7da74a9213b..482059fdf79 100644
--- a/telegram/ext/callbackqueryhandler.py
+++ b/telegram/ext/callbackqueryhandler.py
@@ -45,6 +45,8 @@ class CallbackQueryHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -55,6 +57,10 @@ class CallbackQueryHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
             :attr:`check_update` has determined that an update should be processed by this handler.
@@ -91,6 +97,8 @@ class CallbackQueryHandler(Handler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
@@ -102,13 +110,15 @@ def __init__(self,
                  pass_groups=False,
                  pass_groupdict=False,
                  pass_user_data=False,
-                 pass_chat_data=False):
+                 pass_chat_data=False,
+                 run_async=False):
         super().__init__(
             callback,
             pass_update_queue=pass_update_queue,
             pass_job_queue=pass_job_queue,
             pass_user_data=pass_user_data,
-            pass_chat_data=pass_chat_data)
+            pass_chat_data=pass_chat_data,
+            run_async=run_async)
 
         if isinstance(pattern, str):
             pattern = re.compile(pattern)
diff --git a/telegram/ext/choseninlineresulthandler.py b/telegram/ext/choseninlineresulthandler.py
index 69499e6c7ae..46bee4d2be0 100644
--- a/telegram/ext/choseninlineresulthandler.py
+++ b/telegram/ext/choseninlineresulthandler.py
@@ -35,6 +35,8 @@ class ChosenInlineResultHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -45,6 +47,10 @@ class ChosenInlineResultHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
             :attr:`check_update` has determined that an update should be processed by this handler.
@@ -70,6 +76,8 @@ class ChosenInlineResultHandler(Handler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
diff --git a/telegram/ext/commandhandler.py b/telegram/ext/commandhandler.py
index 5b97f5d9712..c03d04d4cfa 100644
--- a/telegram/ext/commandhandler.py
+++ b/telegram/ext/commandhandler.py
@@ -60,6 +60,8 @@ class CommandHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a :obj:`dict` you
@@ -70,6 +72,10 @@ class CommandHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         command (:obj:`str` | List[:obj:`str`]): The command or list of commands this handler
             should listen for. Limitations are the same as described here
@@ -111,6 +117,8 @@ class CommandHandler(Handler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     Raises:
         ValueError - when command is too long or has illegal chars.
@@ -125,13 +133,15 @@ def __init__(self,
                  pass_update_queue=False,
                  pass_job_queue=False,
                  pass_user_data=False,
-                 pass_chat_data=False):
+                 pass_chat_data=False,
+                 run_async=False):
         super().__init__(
             callback,
             pass_update_queue=pass_update_queue,
             pass_job_queue=pass_job_queue,
             pass_user_data=pass_user_data,
-            pass_chat_data=pass_chat_data)
+            pass_chat_data=pass_chat_data,
+            run_async=run_async)
 
         if isinstance(command, str):
             self.command = [command.lower()]
@@ -242,6 +252,8 @@ class PrefixHandler(CommandHandler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -252,6 +264,10 @@ class PrefixHandler(CommandHandler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         prefix (:obj:`str` | List[:obj:`str`]): The prefix(es) that will precede :attr:`command`.
         command (:obj:`str` | List[:obj:`str`]): The command or list of commands this handler
@@ -289,6 +305,8 @@ class PrefixHandler(CommandHandler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
@@ -301,7 +319,8 @@ def __init__(self,
                  pass_update_queue=False,
                  pass_job_queue=False,
                  pass_user_data=False,
-                 pass_chat_data=False):
+                 pass_chat_data=False,
+                 run_async=False):
 
         self._prefix = list()
         self._command = list()
@@ -312,7 +331,8 @@ def __init__(self,
             pass_update_queue=pass_update_queue,
             pass_job_queue=pass_job_queue,
             pass_user_data=pass_user_data,
-            pass_chat_data=pass_chat_data)
+            pass_chat_data=pass_chat_data,
+            run_async=run_async)
 
         self.prefix = prefix
         self.command = command
diff --git a/telegram/ext/conversationhandler.py b/telegram/ext/conversationhandler.py
index 811d82b10db..1f3b0dc8836 100644
--- a/telegram/ext/conversationhandler.py
+++ b/telegram/ext/conversationhandler.py
@@ -168,6 +168,7 @@ def __init__(self,
                  name=None,
                  persistent=False,
                  map_to_parent=None):
+        self.run_async = False
 
         self._entry_points = entry_points
         self._states = states
diff --git a/telegram/ext/dispatcher.py b/telegram/ext/dispatcher.py
index 828993c53b0..31da7a429e8 100644
--- a/telegram/ext/dispatcher.py
+++ b/telegram/ext/dispatcher.py
@@ -48,13 +48,18 @@ def run_async(func):
     Using this decorator is only possible when only a single Dispatcher exist in the system.
 
     Warning:
-        If you're using @run_async you cannot rely on adding custom attributes to
+        If you're using ``@run_async`` you cannot rely on adding custom attributes to
         :class:`telegram.ext.CallbackContext`. See its docs for more info.
     """
 
     @wraps(func)
     def async_func(*args, **kwargs):
-        return Dispatcher.get_instance().run_async(func, *args, **kwargs)
+        warnings.warn('The @run_async decorator is deprecated. Use the `run_async` parameter of'
+                      '`Dispatcher.add_handler` or `Dispatcher.run_async` instead.',
+                      TelegramDeprecationWarning,
+                      stacklevel=2)
+        return Dispatcher.get_instance().run_async(func, *args, update=None, error_handler=None,
+                                                   **kwargs)
 
     return async_func
 
@@ -91,8 +96,8 @@ class Dispatcher:
         update_queue (:obj:`Queue`): The synchronized queue that will contain the updates.
         job_queue (:class:`telegram.ext.JobQueue`): Optional. The :class:`telegram.ext.JobQueue`
             instance to pass onto handler callbacks.
-        workers (:obj:`int`): Number of maximum concurrent worker threads for the ``@run_async``
-            decorator.
+        workers (:obj:`int`, optional): Number of maximum concurrent worker threads for the
+            ``@run_async`` decorator and :meth:`run_async`.
         user_data (:obj:`defaultdict`): A dictionary handlers can use to store data for the user.
         chat_data (:obj:`defaultdict`): A dictionary handlers can use to store data for the chat.
         bot_data (:obj:`dict`): A dictionary handlers can use to store data for the bot.
@@ -105,7 +110,7 @@ class Dispatcher:
         job_queue (:class:`telegram.ext.JobQueue`, optional): The :class:`telegram.ext.JobQueue`
                 instance to pass onto handler callbacks.
         workers (:obj:`int`, optional): Number of maximum concurrent worker threads for the
-            ``@run_async`` decorator. Defaults to 4.
+            ``@run_async`` decorator and :meth:`run_async`. Defaults to 4.
         persistence (:class:`telegram.ext.BasePersistence`, optional): The persistence class to
             store data that should be persistent over restarts.
         use_context (:obj:`bool`, optional): If set to :obj:`True` Use the context based
@@ -141,6 +146,7 @@ def __init__(self,
         self.user_data = defaultdict(dict)
         self.chat_data = defaultdict(dict)
         self.bot_data = {}
+        self._update_persistence_lock = Lock()
         if persistence:
             if not isinstance(persistence, BasePersistence):
                 raise TypeError("persistence should be based on telegram.ext.BasePersistence")
@@ -164,8 +170,9 @@ def __init__(self,
         """Dict[:obj:`int`, List[:class:`telegram.ext.Handler`]]: Holds the handlers per group."""
         self.groups = []
         """List[:obj:`int`]: A list with all groups."""
-        self.error_handlers = []
-        """List[:obj:`callable`]: A list of errorHandlers."""
+        self.error_handlers = {}
+        """Dict[:obj:`callable`, :obj:`bool`]: A dict, where the keys are error handlers and the
+        values indicate whether they are to be run asynchronously."""
 
         self.running = False
         """:obj:`bool`: Indicates if this dispatcher is running."""
@@ -229,30 +236,84 @@ def _pooled(self):
                 break
 
             promise.run()
+
             if isinstance(promise.exception, DispatcherHandlerStop):
                 self.logger.warning(
                     'DispatcherHandlerStop is not supported with async functions; func: %s',
                     promise.pooled_function.__name__)
+            elif isinstance(promise.exception, Exception):
+                if promise.pooled_function in self.error_handlers:
+                    self.logger.error('An error was raised while processing the update '
+                                      'and an uncaught error was raised while handling '
+                                      'the error with an error_handler')
+                elif promise.error_handler:
+                    try:
+                        context = CallbackContext.from_error(promise.update, promise.exception,
+                                                             self)
+                        if promise.update:
+                            promise.error_handler(promise.update, context, *promise.args,
+                                                  **promise.kwargs)
+                        else:
+                            promise.error_handler(context, *promise.args, **promise.kwargs)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                elif promise.update:
+                    try:
+                        self.dispatch_error(promise.update, promise.exception)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                else:
+                    self.logger.exception(
+                        'No error handlers are set for async function, logging exception.',
+                        exc_info=promise.exception)
+            else:
+                self.update_persistence(update=promise.update)
 
-    def run_async(self, func, *args, **kwargs):
+    def run_async(self, func, *args, update=None, error_handler=None, **kwargs):
         """Queue a function (with given args/kwargs) to be run asynchronously.
 
         Warning:
-            If you're using @run_async you cannot rely on adding custom attributes to
-            :class:`telegram.ext.CallbackContext`. See its docs for more info.
+            * If you're using ``@run_async``/:meth:`run_async` you cannot rely on adding custom
+              attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+            * When using :meth:`run_async` within an asynchronously run error handler (see
+              :meth:`add_error_handler`) and ``update`` is passed, you *must* also pass
+              ``error_handler``, as an exception within ``func`` would cause an infinite loop
+              otherwise.
+            * passing ``error_handler`` is not supported for the old style API, i.e.
+              ``use_context=False``
 
         Args:
             func (:obj:`callable`): The function to run in the thread.
-            *args (:obj:`tuple`, optional): Arguments to `func`.
-            **kwargs (:obj:`dict`, optional): Keyword arguments to `func`.
+            *args (:obj:`tuple`, optional): Arguments to ``func``.
+            update (:class:`telegram.Update`, optional): The update associated with the functions
+                call.
+            error_handler (:obj:`callable`, optional): An error handler to call in case of an
+                exception during the call of ``func``. The callback signature of ``error_handler``
+                must be
+
+                ``def error_handler(context: CallbackContext, *args, **kwargs)``
+
+                or
+
+                ``def error_handler(update: Update, context: CallbackContext, *args, **kwargs)``
+
+                if ``update`` is passed, where ``*args`` and ``**kwargs`` are the arguments for
+                ``func``.
+            **kwargs (:obj:`dict`, optional): Keyword arguments to ``func``.
 
         Returns:
             Promise
 
         """
-        # TODO: handle exception in async threads
-        #       set a threading.Event to notify caller thread
-        promise = Promise(func, args, kwargs)
+        if self.use_context is False and error_handler:
+            raise ValueError('Passing `error_handler` is not supported for non context aware '
+                             'dispatchers!')
+
+        promise = Promise(func, args, kwargs, update=update, error_handler=error_handler)
         self.__async_queue.put(promise)
         return promise
 
@@ -358,7 +419,10 @@ def process_update(self, update):
                         if not context and self.use_context:
                             context = CallbackContext.from_update(update, self)
                         handler.handle_update(update, self, check, context)
-                        self.update_persistence(update=update)
+
+                        # If handler runs async updating immediately doesn't make sense
+                        if not handler.run_async:
+                            self.update_persistence(update=update)
                         break
 
             # Stop processing with any other handler.
@@ -415,7 +479,7 @@ def add_handler(self, handler, group=DEFAULT_GROUP):
         if isinstance(handler, ConversationHandler) and handler.persistent:
             if not self.persistence:
                 raise ValueError(
-                    "Conversationhandler {} can not be persistent if dispatcher has no "
+                    "ConversationHandler {} can not be persistent if dispatcher has no "
                     "persistence".format(handler.name))
             handler.persistence = self.persistence
             handler.conversations = self.persistence.get_conversations(handler.name)
@@ -448,57 +512,58 @@ def update_persistence(self, update=None):
             update (:class:`telegram.Update`, optional): The update to process. If passed, only the
             corresponding ``user_data`` and ``chat_data`` will be updated.
         """
-        if self.persistence:
-            chat_ids = self.chat_data.keys()
-            user_ids = self.user_data.keys()
-
-            if isinstance(update, Update):
-                if update.effective_chat:
-                    chat_ids = [update.effective_chat.id]
-                else:
-                    chat_ids = []
-                if update.effective_user:
-                    user_ids = [update.effective_user.id]
-                else:
-                    user_ids = []
-
-            if self.persistence.store_bot_data:
-                try:
-                    self.persistence.update_bot_data(self.bot_data)
-                except Exception as e:
-                    try:
-                        self.dispatch_error(update, e)
-                    except Exception:
-                        message = 'Saving bot data raised an error and an ' \
-                                  'uncaught error was raised while handling ' \
-                                  'the error with an error_handler'
-                        self.logger.exception(message)
-            if self.persistence.store_chat_data:
-                for chat_id in chat_ids:
+        with self._update_persistence_lock:
+            if self.persistence:
+                chat_ids = list(self.chat_data.keys())
+                user_ids = list(self.user_data.keys())
+
+                if isinstance(update, Update):
+                    if update.effective_chat:
+                        chat_ids = [update.effective_chat.id]
+                    else:
+                        chat_ids = []
+                    if update.effective_user:
+                        user_ids = [update.effective_user.id]
+                    else:
+                        user_ids = []
+
+                if self.persistence.store_bot_data:
                     try:
-                        self.persistence.update_chat_data(chat_id, self.chat_data[chat_id])
+                        self.persistence.update_bot_data(self.bot_data)
                     except Exception as e:
                         try:
                             self.dispatch_error(update, e)
                         except Exception:
-                            message = 'Saving chat data raised an error and an ' \
+                            message = 'Saving bot data raised an error and an ' \
                                       'uncaught error was raised while handling ' \
                                       'the error with an error_handler'
                             self.logger.exception(message)
-            if self.persistence.store_user_data:
-                for user_id in user_ids:
-                    try:
-                        self.persistence.update_user_data(user_id, self.user_data[user_id])
-                    except Exception as e:
+                if self.persistence.store_chat_data:
+                    for chat_id in chat_ids:
                         try:
-                            self.dispatch_error(update, e)
-                        except Exception:
-                            message = 'Saving user data raised an error and an ' \
-                                      'uncaught error was raised while handling ' \
-                                      'the error with an error_handler'
-                            self.logger.exception(message)
-
-    def add_error_handler(self, callback):
+                            self.persistence.update_chat_data(chat_id, self.chat_data[chat_id])
+                        except Exception as e:
+                            try:
+                                self.dispatch_error(update, e)
+                            except Exception:
+                                message = 'Saving chat data raised an error and an ' \
+                                          'uncaught error was raised while handling ' \
+                                          'the error with an error_handler'
+                                self.logger.exception(message)
+                if self.persistence.store_user_data:
+                    for user_id in user_ids:
+                        try:
+                            self.persistence.update_user_data(user_id, self.user_data[user_id])
+                        except Exception as e:
+                            try:
+                                self.dispatch_error(update, e)
+                            except Exception:
+                                message = 'Saving user data raised an error and an ' \
+                                          'uncaught error was raised while handling ' \
+                                          'the error with an error_handler'
+                                self.logger.exception(message)
+
+    def add_error_handler(self, callback, run_async=False):
         """Registers an error handler in the Dispatcher. This handler will receive every error
         which happens in your bot.
 
@@ -512,11 +577,13 @@ def add_error_handler(self, callback):
                 ``def callback(update: Update, context: CallbackContext)``
 
                 The error that happened will be present in context.error.
+            run_async (:obj:`bool`, optional): Whether this handlers callback should be run
+                asynchronously using :meth:`run_async`. Defaults to :obj:`False`.
 
         Note:
             See https://git.io/fxJuV for more info about switching to context based API.
         """
-        self.error_handlers.append(callback)
+        self.error_handlers[callback] = run_async
 
     def remove_error_handler(self, callback):
         """Removes an error handler.
@@ -525,8 +592,7 @@ def remove_error_handler(self, callback):
             callback (:obj:`callable`): The error handler to remove.
 
         """
-        if callback in self.error_handlers:
-            self.error_handlers.remove(callback)
+        self.error_handlers.pop(callback, None)
 
     def dispatch_error(self, update, error):
         """Dispatches an error.
@@ -537,11 +603,20 @@ def dispatch_error(self, update, error):
 
         """
         if self.error_handlers:
-            for callback in self.error_handlers:
-                if self.use_context:
-                    callback(update, CallbackContext.from_error(update, error, self))
+            for callback, run_async in self.error_handlers.items():
+                if run_async:
+                    if self.use_context:
+                        self.run_async(callback,
+                                       update,
+                                       CallbackContext.from_error(update, error, self),
+                                       update=update)
+                    else:
+                        self.run_async(callback, self.bot, update, error, update=update)
                 else:
-                    callback(self.bot, update, error)
+                    if self.use_context:
+                        callback(update, CallbackContext.from_error(update, error, self))
+                    else:
+                        callback(self.bot, update, error)
 
         else:
             self.logger.exception(
diff --git a/telegram/ext/handler.py b/telegram/ext/handler.py
index 8126a8fc582..38bb10d3823 100644
--- a/telegram/ext/handler.py
+++ b/telegram/ext/handler.py
@@ -34,6 +34,8 @@ class Handler(ABC):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -44,6 +46,10 @@ class Handler(ABC):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
             :attr:`check_update` has determined that an update should be processed by this handler.
@@ -69,6 +75,8 @@ class Handler(ABC):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
@@ -77,12 +85,14 @@ def __init__(self,
                  pass_update_queue=False,
                  pass_job_queue=False,
                  pass_user_data=False,
-                 pass_chat_data=False):
+                 pass_chat_data=False,
+                 run_async=False):
         self.callback = callback
         self.pass_update_queue = pass_update_queue
         self.pass_job_queue = pass_job_queue
         self.pass_user_data = pass_user_data
         self.pass_chat_data = pass_chat_data
+        self.run_async = run_async
 
     @abstractmethod
     def check_update(self, update):
@@ -116,10 +126,17 @@ def handle_update(self, update, dispatcher, check_result, context=None):
         """
         if context:
             self.collect_additional_context(context, update, dispatcher, check_result)
-            return self.callback(update, context)
+            if self.run_async:
+                return dispatcher.run_async(self.callback, update, context, update=update)
+            else:
+                return self.callback(update, context)
         else:
             optional_args = self.collect_optional_args(dispatcher, update, check_result)
-            return self.callback(dispatcher.bot, update, **optional_args)
+            if self.run_async:
+                return dispatcher.run_async(self.callback, dispatcher.bot, update, update=update,
+                                            **optional_args)
+            else:
+                return self.callback(dispatcher.bot, update, **optional_args)
 
     def collect_additional_context(self, context, update, dispatcher, check_result):
         """Prepares additional arguments for the context. Override if needed.
diff --git a/telegram/ext/inlinequeryhandler.py b/telegram/ext/inlinequeryhandler.py
index 39612b5629b..b898bcac109 100644
--- a/telegram/ext/inlinequeryhandler.py
+++ b/telegram/ext/inlinequeryhandler.py
@@ -45,6 +45,8 @@ class InlineQueryHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -55,6 +57,10 @@ class InlineQueryHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
             :attr:`check_update` has determined that an update should be processed by this handler.
@@ -91,6 +97,8 @@ class InlineQueryHandler(Handler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
@@ -102,13 +110,15 @@ def __init__(self,
                  pass_groups=False,
                  pass_groupdict=False,
                  pass_user_data=False,
-                 pass_chat_data=False):
+                 pass_chat_data=False,
+                 run_async=False):
         super().__init__(
             callback,
             pass_update_queue=pass_update_queue,
             pass_job_queue=pass_job_queue,
             pass_user_data=pass_user_data,
-            pass_chat_data=pass_chat_data)
+            pass_chat_data=pass_chat_data,
+            run_async=False)
 
         if isinstance(pattern, str):
             pattern = re.compile(pattern)
diff --git a/telegram/ext/messagehandler.py b/telegram/ext/messagehandler.py
index b0d4d8b5cf8..6e29d1dbcbd 100644
--- a/telegram/ext/messagehandler.py
+++ b/telegram/ext/messagehandler.py
@@ -48,6 +48,8 @@ class MessageHandler(Handler):
             Default is :obj:`None`.
         edited_updates (:obj:`bool`): Should "edited" message updates be handled?
             Default is :obj:`None`.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -58,6 +60,10 @@ class MessageHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         filters (:class:`telegram.ext.BaseFilter`, optional): A filter inheriting from
             :class:`telegram.ext.filters.BaseFilter`. Standard filters can be found in
@@ -100,6 +106,8 @@ class MessageHandler(Handler):
         edited_updates (:obj:`bool`, optional): Should "edited" message updates be handled? Default
             is :obj:`None`.
             DEPRECATED: Please switch to filters for update filtering.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     Raises:
         ValueError
@@ -115,14 +123,16 @@ def __init__(self,
                  pass_chat_data=False,
                  message_updates=None,
                  channel_post_updates=None,
-                 edited_updates=None):
+                 edited_updates=None,
+                 run_async=False):
 
         super().__init__(
             callback,
             pass_update_queue=pass_update_queue,
             pass_job_queue=pass_job_queue,
             pass_user_data=pass_user_data,
-            pass_chat_data=pass_chat_data)
+            pass_chat_data=pass_chat_data,
+            run_async=run_async)
         if message_updates is False and channel_post_updates is False and edited_updates is False:
             raise ValueError(
                 'message_updates, channel_post_updates and edited_updates are all False')
diff --git a/telegram/ext/pollanswerhandler.py b/telegram/ext/pollanswerhandler.py
index 86132acd568..6e8d6325793 100644
--- a/telegram/ext/pollanswerhandler.py
+++ b/telegram/ext/pollanswerhandler.py
@@ -34,6 +34,8 @@ class PollAnswerHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -44,6 +46,10 @@ class PollAnswerHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
             :attr:`check_update` has determined that an update should be processed by this handler.
@@ -69,6 +75,8 @@ class PollAnswerHandler(Handler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
diff --git a/telegram/ext/pollhandler.py b/telegram/ext/pollhandler.py
index 4ff191b8d2c..b3198cd8640 100644
--- a/telegram/ext/pollhandler.py
+++ b/telegram/ext/pollhandler.py
@@ -34,6 +34,8 @@ class PollHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -44,6 +46,10 @@ class PollHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
             :attr:`check_update` has determined that an update should be processed by this handler.
@@ -69,6 +75,8 @@ class PollHandler(Handler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
diff --git a/telegram/ext/precheckoutqueryhandler.py b/telegram/ext/precheckoutqueryhandler.py
index 0d3b7314c80..cf381332a30 100644
--- a/telegram/ext/precheckoutqueryhandler.py
+++ b/telegram/ext/precheckoutqueryhandler.py
@@ -35,6 +35,8 @@ class PreCheckoutQueryHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -45,6 +47,10 @@ class PreCheckoutQueryHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
             :attr:`check_update` has determined that an update should be processed by this handler.
@@ -70,6 +76,8 @@ class PreCheckoutQueryHandler(Handler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
diff --git a/telegram/ext/regexhandler.py b/telegram/ext/regexhandler.py
index 4921f007dc5..1a9e5bc8b97 100644
--- a/telegram/ext/regexhandler.py
+++ b/telegram/ext/regexhandler.py
@@ -48,11 +48,17 @@ class RegexHandler(MessageHandler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         This handler is being deprecated. For the same use case use:
         ``MessageHandler(Filters.regex(r'pattern'), callback)``
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
 
     Args:
         pattern (:obj:`str` | :obj:`Pattern`): The regex pattern.
@@ -88,6 +94,8 @@ class RegexHandler(MessageHandler):
             Default is :obj:`True`.
         edited_updates (:obj:`bool`, optional): Should "edited" message updates be handled? Default
             is :obj:`False`.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     Raises:
         ValueError
@@ -106,7 +114,8 @@ def __init__(self,
                  allow_edited=False,
                  message_updates=True,
                  channel_post_updates=False,
-                 edited_updates=False):
+                 edited_updates=False,
+                 run_async=False):
         warnings.warn('RegexHandler is deprecated. See https://git.io/fxJuV for more info',
                       TelegramDeprecationWarning,
                       stacklevel=2)
@@ -118,7 +127,8 @@ def __init__(self,
                          pass_chat_data=pass_chat_data,
                          message_updates=message_updates,
                          channel_post_updates=channel_post_updates,
-                         edited_updates=edited_updates)
+                         edited_updates=edited_updates,
+                         run_async=run_async)
         self.pass_groups = pass_groups
         self.pass_groupdict = pass_groupdict
 
diff --git a/telegram/ext/shippingqueryhandler.py b/telegram/ext/shippingqueryhandler.py
index becc976c2ad..5ecb1e968bc 100644
--- a/telegram/ext/shippingqueryhandler.py
+++ b/telegram/ext/shippingqueryhandler.py
@@ -35,6 +35,8 @@ class ShippingQueryHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Note:
         :attr:`pass_user_data` and :attr:`pass_chat_data` determine whether a ``dict`` you
@@ -45,6 +47,10 @@ class ShippingQueryHandler(Handler):
         Note that this is DEPRECATED, and you should use context based callbacks. See
         https://git.io/fxJuV for more info.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
             :attr:`check_update` has determined that an update should be processed by this handler.
@@ -70,6 +76,8 @@ class ShippingQueryHandler(Handler):
         pass_chat_data (:obj:`bool`, optional): If set to :obj:`True`, a keyword argument called
             ``chat_data`` will be passed to the callback function. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
diff --git a/telegram/ext/stringcommandhandler.py b/telegram/ext/stringcommandhandler.py
index dc2e4753665..0ea6678db74 100644
--- a/telegram/ext/stringcommandhandler.py
+++ b/telegram/ext/stringcommandhandler.py
@@ -28,6 +28,10 @@ class StringCommandHandler(Handler):
         This handler is not used to handle Telegram :attr:`telegram.Update`, but strings manually
         put in the queue. For example to send messages with the bot using command line or API.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Attributes:
         command (:obj:`str`): The command this handler should listen for.
         callback (:obj:`callable`): The callback function for this handler.
@@ -37,6 +41,8 @@ class StringCommandHandler(Handler):
             passed to the callback function.
         pass_job_queue (:obj:`bool`): Determines whether ``job_queue`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Args:
         callback (:obj:`callable`): The callback function for this handler. Will be called when
@@ -62,6 +68,8 @@ class StringCommandHandler(Handler):
             class:`telegram.ext.JobQueue` instance created by the :class:`telegram.ext.Updater`
             which can be used to schedule new jobs. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
@@ -70,11 +78,13 @@ def __init__(self,
                  callback,
                  pass_args=False,
                  pass_update_queue=False,
-                 pass_job_queue=False):
+                 pass_job_queue=False,
+                 run_async=False):
         super().__init__(
             callback,
             pass_update_queue=pass_update_queue,
-            pass_job_queue=pass_job_queue)
+            pass_job_queue=pass_job_queue,
+            run_async=run_async)
         self.command = command
         self.pass_args = pass_args
 
diff --git a/telegram/ext/stringregexhandler.py b/telegram/ext/stringregexhandler.py
index ff1716470e1..d88980e07ba 100644
--- a/telegram/ext/stringregexhandler.py
+++ b/telegram/ext/stringregexhandler.py
@@ -33,6 +33,10 @@ class StringRegexHandler(Handler):
         This handler is not used to handle Telegram :attr:`telegram.Update`, but strings manually
         put in the queue. For example to send messages with the bot using command line or API.
 
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+
     Attributes:
         pattern (:obj:`str` | :obj:`Pattern`): The regex pattern.
         callback (:obj:`callable`): The callback function for this handler.
@@ -44,6 +48,8 @@ class StringRegexHandler(Handler):
             passed to the callback function.
         pass_job_queue (:obj:`bool`): Determines whether ``job_queue`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
 
     Args:
         pattern (:obj:`str` | :obj:`Pattern`): The regex pattern.
@@ -73,6 +79,8 @@ class StringRegexHandler(Handler):
             :class:`telegram.ext.JobQueue` instance created by the :class:`telegram.ext.Updater`
             which can be used to schedule new jobs. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
@@ -82,11 +90,13 @@ def __init__(self,
                  pass_groups=False,
                  pass_groupdict=False,
                  pass_update_queue=False,
-                 pass_job_queue=False):
+                 pass_job_queue=False,
+                 run_async=False):
         super().__init__(
             callback,
             pass_update_queue=pass_update_queue,
-            pass_job_queue=pass_job_queue)
+            pass_job_queue=pass_job_queue,
+            run_async=run_async)
 
         if isinstance(pattern, str):
             pattern = re.compile(pattern)
diff --git a/telegram/ext/typehandler.py b/telegram/ext/typehandler.py
index 3bb5e990b3f..ed40c1561de 100644
--- a/telegram/ext/typehandler.py
+++ b/telegram/ext/typehandler.py
@@ -32,6 +32,12 @@ class TypeHandler(Handler):
             passed to the callback function.
         pass_job_queue (:obj:`bool`): Determines whether ``job_queue`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
+
+    Warning:
+        When setting ``run_async`` to :obj:`True`, you cannot rely on adding custom
+        attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
 
     Args:
         type (:obj:`type`): The ``type`` of updates this handler should process, as
@@ -56,6 +62,8 @@ class TypeHandler(Handler):
             :class:`telegram.ext.JobQueue` instance created by the :class:`telegram.ext.Updater`
             which can be used to schedule new jobs. Default is :obj:`False`.
             DEPRECATED: Please switch to context based callbacks.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously. Defaults to :obj:`False`.
 
     """
 
@@ -64,11 +72,13 @@ def __init__(self,
                  callback,
                  strict=False,
                  pass_update_queue=False,
-                 pass_job_queue=False):
+                 pass_job_queue=False,
+                 run_async=False):
         super().__init__(
             callback,
             pass_update_queue=pass_update_queue,
-            pass_job_queue=pass_job_queue)
+            pass_job_queue=pass_job_queue,
+            run_async=run_async)
         self.type = type
         self.strict = strict
 
diff --git a/telegram/utils/promise.py b/telegram/utils/promise.py
index 76139a6f138..ee65fab8602 100644
--- a/telegram/utils/promise.py
+++ b/telegram/utils/promise.py
@@ -41,10 +41,12 @@ class Promise:
 
     """
 
-    def __init__(self, pooled_function, args, kwargs):
+    def __init__(self, pooled_function, args, kwargs, update=None, error_handler=None):
         self.pooled_function = pooled_function
         self.args = args
         self.kwargs = kwargs
+        self.update = update
+        self.error_handler = error_handler
         self.done = Event()
         self._result = None
         self._exception = None
@@ -56,7 +58,6 @@ def run(self):
             self._result = self.pooled_function(*self.args, **self.kwargs)
 
         except Exception as exc:
-            logger.exception('An uncaught error was raised while running the promise')
             self._exception = exc
 
         finally:
diff --git a/tests/conftest.py b/tests/conftest.py
index e6423476e55..5f00fa5f61b 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -120,7 +120,7 @@ def dp(_dp):
     _dp.persistence = None
     _dp.handlers = {}
     _dp.groups = []
-    _dp.error_handlers = []
+    _dp.error_handlers = {}
     _dp.__stop_event = Event()
     _dp.__exception_event = Event()
     _dp.__async_queue = Queue()
diff --git a/tests/test_conversationhandler.py b/tests/test_conversationhandler.py
index 4d452f3d028..936cb4c3fec 100644
--- a/tests/test_conversationhandler.py
+++ b/tests/test_conversationhandler.py
@@ -486,6 +486,30 @@ def test_end_on_first_message_async(self, dp, bot, user1):
         # Assert that the Promise has been resolved and the conversation ended.
         assert len(handler.conversations) == 0
 
+    def test_end_on_first_message_async_handler(self, dp, bot, user1):
+        handler = ConversationHandler(
+            entry_points=[CommandHandler('start', self.start_end, run_async=True)], states={},
+            fallbacks=[])
+        dp.add_handler(handler)
+
+        # User starts the state machine with an async function that immediately ends the
+        # conversation. Async results are resolved when the users state is queried next time.
+        message = Message(0, user1, None, self.group, text='/start',
+                          entities=[MessageEntity(type=MessageEntity.BOT_COMMAND,
+                                                  offset=0, length=len('/start'))],
+                          bot=bot)
+        dp.update_queue.put(Update(update_id=0, message=message))
+        sleep(.1)
+        # Assert that the Promise has been accepted as the new state
+        assert len(handler.conversations) == 1
+
+        message.text = 'resolve promise pls'
+        message.entities[0].length = len('resolve promise pls')
+        dp.update_queue.put(Update(update_id=0, message=message))
+        sleep(.1)
+        # Assert that the Promise has been resolved and the conversation ended.
+        assert len(handler.conversations) == 0
+
     def test_none_on_first_message(self, dp, bot, user1):
         handler = ConversationHandler(
             entry_points=[CommandHandler('start', self.start_none)], states={}, fallbacks=[])
@@ -520,6 +544,29 @@ def test_none_on_first_message_async(self, dp, bot, user1):
         # Assert that the Promise has been resolved and the conversation ended.
         assert len(handler.conversations) == 0
 
+    def test_none_on_first_message_async_handler(self, dp, bot, user1):
+        handler = ConversationHandler(
+            entry_points=[CommandHandler('start', self.start_none, run_async=True)], states={},
+            fallbacks=[])
+        dp.add_handler(handler)
+
+        # User starts the state machine with an async function that returns None
+        # Async results are resolved when the users state is queried next time.
+        message = Message(0, user1, None, self.group, text='/start',
+                          entities=[MessageEntity(type=MessageEntity.BOT_COMMAND,
+                                                  offset=0, length=len('/start'))],
+                          bot=bot)
+        dp.update_queue.put(Update(update_id=0, message=message))
+        sleep(.1)
+        # Assert that the Promise has been accepted as the new state
+        assert len(handler.conversations) == 1
+
+        message.text = 'resolve promise pls'
+        dp.update_queue.put(Update(update_id=0, message=message))
+        sleep(.1)
+        # Assert that the Promise has been resolved and the conversation ended.
+        assert len(handler.conversations) == 0
+
     def test_per_chat_message_without_chat(self, bot, user1):
         handler = ConversationHandler(
             entry_points=[CommandHandler('start', self.start_end)], states={},
diff --git a/tests/test_dispatcher.py b/tests/test_dispatcher.py
index 26949ddc5dd..7c95c969058 100644
--- a/tests/test_dispatcher.py
+++ b/tests/test_dispatcher.py
@@ -16,6 +16,7 @@
 #
 # You should have received a copy of the GNU Lesser Public License
 # along with this program.  If not, see [http://www.gnu.org/licenses/].
+import logging
 from queue import Queue
 from threading import current_thread
 from time import sleep
@@ -54,6 +55,9 @@ def reset(self):
     def error_handler(self, bot, update, error):
         self.received = error.message
 
+    def error_handler_context(self, update, context):
+        self.received = context.error.message
+
     def error_handler_raise_error(self, bot, update, error):
         raise Exception('Failing bigly')
 
@@ -67,7 +71,9 @@ def callback(bot, update):
         return callback
 
     def callback_raise_error(self, bot, update):
-        raise TelegramError(update.message.text)
+        if isinstance(bot, Bot):
+            raise TelegramError(update.message.text)
+        raise TelegramError(bot.message.text)
 
     def callback_if_not_update_queue(self, bot, update, update_queue=None):
         if update_queue is not None:
@@ -190,6 +196,166 @@ def test_run_async_with_args(self, dp):
         sleep(.1)
         assert self.received == self.message_update.message
 
+    def test_multiple_run_async_deprecation(self, dp):
+        assert isinstance(dp, Dispatcher)
+
+        @run_async
+        def callback(update, context):
+            pass
+
+        dp.add_handler(MessageHandler(Filters.all, callback))
+
+        with pytest.warns(TelegramDeprecationWarning, match='@run_async decorator'):
+            dp.process_update(self.message_update)
+
+    def test_async_raises_dispatcher_handler_stop(self, dp, caplog):
+        @run_async
+        def callback(update, context):
+            raise DispatcherHandlerStop()
+
+        dp.add_handler(MessageHandler(Filters.all, callback))
+
+        with caplog.at_level(logging.WARNING):
+            dp.update_queue.put(self.message_update)
+            sleep(.1)
+            assert len(caplog.records) == 1
+            assert caplog.records[-1].msg.startswith('DispatcherHandlerStop is not supported '
+                                                     'with async functions')
+
+    def test_add_async_handler(self, dp):
+        dp.add_handler(MessageHandler(Filters.all,
+                                      self.callback_if_not_update_queue,
+                                      pass_update_queue=True,
+                                      run_async=True))
+
+        dp.update_queue.put(self.message_update)
+        sleep(.1)
+        assert self.received == self.message_update.message
+
+    def test_run_async_no_error_handler(self, dp, caplog):
+        def func():
+            raise RuntimeError('Async Error')
+
+        with caplog.at_level(logging.ERROR):
+            dp.run_async(func)
+            sleep(.1)
+            assert len(caplog.records) == 1
+            assert caplog.records[-1].msg.startswith('No error handlers are set for async')
+
+    def test_async_handler_error_handler(self, dp):
+        dp.add_handler(MessageHandler(Filters.all,
+                                      self.callback_raise_error,
+                                      run_async=True))
+        dp.add_error_handler(self.error_handler)
+
+        dp.update_queue.put(self.message_update)
+        sleep(.1)
+        assert self.received == self.message_update.message.text
+
+    def test_async_handler_async_error_handler_context(self, cdp):
+        cdp.add_handler(MessageHandler(Filters.all,
+                                       self.callback_raise_error,
+                                       run_async=True))
+        cdp.add_error_handler(self.error_handler_context, run_async=True)
+
+        cdp.update_queue.put(self.message_update)
+        sleep(2)
+        assert self.received == self.message_update.message.text
+
+    def test_async_handler_error_handler_that_raises_error(self, dp, caplog):
+        handler = MessageHandler(Filters.all,
+                                 self.callback_raise_error,
+                                 run_async=True)
+        dp.add_handler(handler)
+        dp.add_error_handler(self.error_handler_raise_error, run_async=False)
+
+        with caplog.at_level(logging.ERROR):
+            dp.update_queue.put(self.message_update)
+            sleep(.1)
+            assert len(caplog.records) == 1
+            assert caplog.records[-1].msg.startswith('An error was raised while processing the')
+
+        # Make sure that the main loop still runs
+        dp.remove_handler(handler)
+        dp.add_handler(MessageHandler(Filters.all,
+                                      self.callback_increase_count,
+                                      run_async=True))
+        dp.update_queue.put(self.message_update)
+        sleep(.1)
+        assert self.count == 1
+
+    def test_async_handler_async_error_handler_that_raises_error(self, dp, caplog):
+        handler = MessageHandler(Filters.all,
+                                 self.callback_raise_error,
+                                 run_async=True)
+        dp.add_handler(handler)
+        dp.add_error_handler(self.error_handler_raise_error, run_async=True)
+
+        with caplog.at_level(logging.ERROR):
+            dp.update_queue.put(self.message_update)
+            sleep(.1)
+            assert len(caplog.records) == 1
+            assert caplog.records[-1].msg.startswith('An error was raised while processing the')
+
+        # Make sure that the main loop still runs
+        dp.remove_handler(handler)
+        dp.add_handler(MessageHandler(Filters.all,
+                                      self.callback_increase_count,
+                                      run_async=True))
+        dp.update_queue.put(self.message_update)
+        sleep(.1)
+        assert self.count == 1
+
+    def test_custom_error_handler_non_context(self, dp):
+        with pytest.raises(ValueError, match='Passing `error_handler` is not'):
+            dp.run_async(lambda x: x, self.message_update, error_handler=lambda x: x)
+
+    def test_custom_error_handler_1(self, cdp):
+        def func(update, b, c=0, d=0):
+            raise RuntimeError('Error')
+
+        def error_handler(update, context, a, b, c=0, d=0):
+            self.received = (isinstance(context, CallbackContext)
+                             and update is self.message_update
+                             and a is self.message_update
+                             and b == 'b'
+                             and c == 'c'
+                             and d == 'd')
+
+        cdp.run_async(func, self.message_update, 'b', update=self.message_update,
+                      error_handler=error_handler, c='c', d='d')
+        sleep(0.1)
+        assert self.received is True
+
+    def test_custom_error_handler_2(self, cdp):
+        def func(update, b, c=0, d=0):
+            raise RuntimeError('Error')
+
+        def error_handler(context, a, b, c=0, d=0):
+            self.received = (isinstance(context, CallbackContext)
+                             and a is self.message_update
+                             and b == 'b'
+                             and c == 'c'
+                             and d == 'd')
+
+        cdp.run_async(func, self.message_update, 'b', error_handler=error_handler, c='c', d='d')
+        sleep(0.1)
+        assert self.received is True
+
+    def test_custom_error_handler_that_raises_error(self, cdp, caplog):
+        def func(update, b, c=0, d=0):
+            raise RuntimeError('Error')
+
+        def error_handler(context, a, b, c=0, d=0):
+            raise TelegramError('ErrorHandler Error')
+
+        with caplog.at_level(logging.ERROR):
+            cdp.run_async(func, self.message_update, 'b', error_handler=error_handler,
+                          c='c', d='d')
+            sleep(.1)
+            assert len(caplog.records) == 1
+            assert caplog.records[-1].msg.startswith('An error was raised while processing the')
+
     def test_error_in_handler(self, dp):
         dp.add_handler(MessageHandler(Filters.all, self.callback_raise_error))
         dp.add_error_handler(self.error_handler)
diff --git a/tests/test_persistence.py b/tests/test_persistence.py
index eb63f7d7cdd..374122e57f4 100644
--- a/tests/test_persistence.py
+++ b/tests/test_persistence.py
@@ -253,6 +253,9 @@ def callback_unknown_user_or_chat(update, context):
         rec = caplog.records[-2]
         assert rec.msg == 'No error handlers are registered, logging exception.'
         assert rec.levelname == 'ERROR'
+        rec = caplog.records[-3]
+        assert rec.msg == 'No error handlers are registered, logging exception.'
+        assert rec.levelname == 'ERROR'
         m.from_user = user2
         m.chat = chat1
         u = Update(1, m)
@@ -281,6 +284,103 @@ def save_user_data(data):
         assert dp.chat_data[-987654][2] == 'test8'
         assert dp.bot_data['test0'] == 'test0'
 
+    def test_dispatcher_integration_handlers_run_async(self, cdp, caplog, bot, base_persistence,
+                                                       chat_data, user_data, bot_data):
+        def get_user_data():
+            return user_data
+
+        def get_chat_data():
+            return chat_data
+
+        def get_bot_data():
+            return bot_data
+
+        base_persistence.get_user_data = get_user_data
+        base_persistence.get_chat_data = get_chat_data
+        base_persistence.get_bot_data = get_bot_data
+        cdp.persistence = base_persistence
+        cdp.user_data = user_data
+        cdp.chat_data = chat_data
+        cdp.bot_data = bot_data
+
+        def callback_known_user(update, context):
+            if not context.user_data['test1'] == 'test2':
+                pytest.fail('user_data corrupt')
+            if not context.bot_data == bot_data:
+                pytest.fail('bot_data corrupt')
+
+        def callback_known_chat(update, context):
+            if not context.chat_data['test3'] == 'test4':
+                pytest.fail('chat_data corrupt')
+            if not context.bot_data == bot_data:
+                pytest.fail('bot_data corrupt')
+
+        def callback_unknown_user_or_chat(update, context):
+            if not context.user_data == {}:
+                pytest.fail('user_data corrupt')
+            if not context.chat_data == {}:
+                pytest.fail('chat_data corrupt')
+            if not context.bot_data == bot_data:
+                pytest.fail('bot_data corrupt')
+            context.user_data[1] = 'test7'
+            context.chat_data[2] = 'test8'
+            context.bot_data['test0'] = 'test0'
+
+        known_user = MessageHandler(Filters.user(user_id=12345), callback_known_user,
+                                    pass_chat_data=True, pass_user_data=True, run_async=True)
+        known_chat = MessageHandler(Filters.chat(chat_id=-67890), callback_known_chat,
+                                    pass_chat_data=True, pass_user_data=True, run_async=True)
+        unknown = MessageHandler(Filters.all, callback_unknown_user_or_chat, pass_chat_data=True,
+                                 pass_user_data=True, run_async=True)
+        cdp.add_handler(known_user)
+        cdp.add_handler(known_chat)
+        cdp.add_handler(unknown)
+        user1 = User(id=12345, first_name='test user', is_bot=False)
+        user2 = User(id=54321, first_name='test user', is_bot=False)
+        chat1 = Chat(id=-67890, type='group')
+        chat2 = Chat(id=-987654, type='group')
+        m = Message(1, user1, None, chat2)
+        u = Update(0, m)
+        with caplog.at_level(logging.ERROR):
+            cdp.process_update(u)
+
+        sleep(.1)
+        rec = caplog.records[-1]
+        assert rec.msg == 'No error handlers are registered, logging exception.'
+        assert rec.levelname == 'ERROR'
+        rec = caplog.records[-2]
+        assert rec.msg == 'No error handlers are registered, logging exception.'
+        assert rec.levelname == 'ERROR'
+        m.from_user = user2
+        m.chat = chat1
+        u = Update(1, m)
+        cdp.process_update(u)
+        m.chat = chat2
+        u = Update(2, m)
+
+        def save_bot_data(data):
+            if 'test0' not in data:
+                pytest.fail()
+
+        def save_chat_data(data):
+            if -987654 not in data:
+                pytest.fail()
+
+        def save_user_data(data):
+            if 54321 not in data:
+                pytest.fail()
+
+        base_persistence.update_chat_data = save_chat_data
+        base_persistence.update_user_data = save_user_data
+        base_persistence.update_bot_data = save_bot_data
+        cdp.process_update(u)
+
+        sleep(0.1)
+
+        assert cdp.user_data[54321][1] == 'test7'
+        assert cdp.chat_data[-987654][2] == 'test8'
+        assert cdp.bot_data['test0'] == 'test0'
+
     def test_persistence_dispatcher_arbitrary_update_types(self, dp, base_persistence, caplog):
         # Updates used with TypeHandler doesn't necessarily have the proper attributes for
         # persistence, makes sure it works anyways


Code review comments:
diff_hunk: @@ -229,30 +236,84 @@ def _pooled(self):
                 break
 
             promise.run()
+
             if isinstance(promise.exception, DispatcherHandlerStop):
                 self.logger.warning(
                     'DispatcherHandlerStop is not supported with async functions; func: %s',
                     promise.pooled_function.__name__)
+            elif isinstance(promise.exception, Exception):
path: telegram/ext/dispatcher.py
text: @user1:
```suggestion
            elif promise.exception:
```

diff_hunk: @@ -229,30 +236,84 @@ def _pooled(self):
                 break
 
             promise.run()
+
             if isinstance(promise.exception, DispatcherHandlerStop):
                 self.logger.warning(
                     'DispatcherHandlerStop is not supported with async functions; func: %s',
                     promise.pooled_function.__name__)
+            elif isinstance(promise.exception, Exception):
+                if promise.pooled_function in self.error_handlers:
+                    self.logger.error('An error was raised while processing the update '
+                                      'and an uncaught error was raised while handling '
+                                      'the error with an error_handler')
+                elif promise.error_handler:
+                    try:
+                        context = CallbackContext.from_error(promise.update, promise.exception,
+                                                             self)
+                        if promise.update:
+                            promise.error_handler(promise.update, context, *promise.args,
+                                                  **promise.kwargs)
+                        else:
+                            promise.error_handler(context, *promise.args, **promise.kwargs)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                elif promise.update:
+                    try:
+                        self.dispatch_error(promise.update, promise.exception)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                else:
+                    self.logger.exception(
+                        'No error handlers are set for async function, logging exception.',
+                        exc_info=promise.exception)
+            else:
+                self.update_persistence(update=promise.update)
path: telegram/ext/dispatcher.py
text: @user1:
I would make the entire function more of "get out of here quickly":
1. Handle the case of happy-path (this must always be the first!)
2. Handle the case of no error handler
3. Handle the case of error handler exists

```
if not promise.exception:
    return self.update_persistence(update=promise.update)

if isinstance(promise.exception, DispatcherHandlerStop):
   return self.logger.warning(....)

if promise.pooled_function in self.error_handlers:
    return self.logger.error('An uncaught error was raised by an error_handler')

# see all `return` above - quick return to handle simple cases.
# and now that we're on the safe zone, we can handle errors
...
```


@author:
That is so much better than the current state o.0

@author:
Uh, except we musn't use `return` as it will end the `while 1` loop. I used continue instead

diff_hunk: @@ -229,30 +236,84 @@ def _pooled(self):
                 break
 
             promise.run()
+
             if isinstance(promise.exception, DispatcherHandlerStop):
                 self.logger.warning(
                     'DispatcherHandlerStop is not supported with async functions; func: %s',
                     promise.pooled_function.__name__)
+            elif isinstance(promise.exception, Exception):
+                if promise.pooled_function in self.error_handlers:
+                    self.logger.error('An error was raised while processing the update '
+                                      'and an uncaught error was raised while handling '
+                                      'the error with an error_handler')
+                elif promise.error_handler:
path: telegram/ext/dispatcher.py
text: @user1:
Now I get confused (probably because of the missing docs on `Promise`):

1. Why is there a `promise.error_handler`: when is it used and why?
2. What's the difference between having `promise.update` and not having one?


@author:
> 1. Why is there a `promise.error_handler`: when is it used and why?

`run_async` can not only be used for running handler callbacks asyncronously, but also custom functions. Custom functions will usually have a other arguments than `update, context`, so it might make sense to have special error handlers form them. So the use case is

```python
def custom_func(a, b, c=None):
   raise RuntimeError()

def error_handler(update, context, a, b, c=None):
   print(f'The error {context.error} happend while running custom_func with the arguments {a}, {b}, {c} for update {update}')

def my_callback(update, context):
    # do some stuff
    context.dispatcher.run_async(custom_func, a, b, update=update, error_handler=error_handler, c=c)
```

> 2. What's the difference between having `promise.update` and not having one?

I was gonna argue about distinguishing handler callbacks from custom functions but acutally I see that it's not really straightforward. I think it would be good to restructure and I have two options in mind:

1. Still allow to set `error_handler`, which can be helpful for custom functions. But the signature can just always be 
    ```python
    def error_handler(update, context, *args, **kwargs):
      ...
    ```
    in case `update` is not passed to `run_async`, it will just be `None` (as is already the case for the usual error handlers called on errors happening in jobs)
    If no `error_handler` is passed, use the dispatchers usual error handlers with the downside of not having access to the parameters of the function that raised the error
2. Remove `error_handler` all together and just use the usual error_handlers. To allow acces to the pooled functions parameters, we could add them to the `context` argument (`context.error_(kw)args`). Maybe even the complete promise (`context.promise`)

Looking at the changes with a bit of distance, I'd acutally go for 2.

Still, for custom functions, error handlers don't necessarily need the `update` parameter, so it's still optional

diff_hunk: @@ -229,30 +236,84 @@ def _pooled(self):
                 break
 
             promise.run()
+
             if isinstance(promise.exception, DispatcherHandlerStop):
                 self.logger.warning(
                     'DispatcherHandlerStop is not supported with async functions; func: %s',
                     promise.pooled_function.__name__)
+            elif isinstance(promise.exception, Exception):
+                if promise.pooled_function in self.error_handlers:
+                    self.logger.error('An error was raised while processing the update '
+                                      'and an uncaught error was raised while handling '
+                                      'the error with an error_handler')
+                elif promise.error_handler:
+                    try:
+                        context = CallbackContext.from_error(promise.update, promise.exception,
+                                                             self)
+                        if promise.update:
+                            promise.error_handler(promise.update, context, *promise.args,
+                                                  **promise.kwargs)
+                        else:
+                            promise.error_handler(context, *promise.args, **promise.kwargs)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                elif promise.update:
+                    try:
+                        self.dispatch_error(promise.update, promise.exception)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                else:
+                    self.logger.exception(
+                        'No error handlers are set for async function, logging exception.',
+                        exc_info=promise.exception)
+            else:
+                self.update_persistence(update=promise.update)
 
-    def run_async(self, func, *args, **kwargs):
+    def run_async(self, func, *args, update=None, error_handler=None, **kwargs):
         """Queue a function (with given args/kwargs) to be run asynchronously.
 
         Warning:
-            If you're using @run_async you cannot rely on adding custom attributes to
-            :class:`telegram.ext.CallbackContext`. See its docs for more info.
+            * If you're using ``@run_async``/:meth:`run_async` you cannot rely on adding custom
+              attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+            * When using :meth:`run_async` within an asynchronously run error handler (see
+              :meth:`add_error_handler`) and ``update`` is passed, you *must* also pass
+              ``error_handler``, as an exception within ``func`` would cause an infinite loop
+              otherwise.
+            * passing ``error_handler`` is not supported for the old style API, i.e.
path: telegram/ext/dispatcher.py
text: @user1:
Maybe remove support for old style API and be done with it?
It makes the code more complicated and thus fragile.

@author:
https://t.me/c/1494805131/5984
https://t.me/c/1494805131/5987
https://t.me/c/1494805131/6006

@user1:
Ok. I see your point.
It's just that seeing how cumbersome the code had become made me second thoughts. But we can keep with this for one more release.

@author:
that's what "passing ``error_handler`` is not supported for the old style API" means?

diff_hunk: @@ -229,30 +236,84 @@ def _pooled(self):
                 break
 
             promise.run()
+
             if isinstance(promise.exception, DispatcherHandlerStop):
                 self.logger.warning(
                     'DispatcherHandlerStop is not supported with async functions; func: %s',
                     promise.pooled_function.__name__)
+            elif isinstance(promise.exception, Exception):
+                if promise.pooled_function in self.error_handlers:
+                    self.logger.error('An error was raised while processing the update '
+                                      'and an uncaught error was raised while handling '
+                                      'the error with an error_handler')
+                elif promise.error_handler:
+                    try:
+                        context = CallbackContext.from_error(promise.update, promise.exception,
+                                                             self)
+                        if promise.update:
+                            promise.error_handler(promise.update, context, *promise.args,
+                                                  **promise.kwargs)
+                        else:
+                            promise.error_handler(context, *promise.args, **promise.kwargs)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                elif promise.update:
+                    try:
+                        self.dispatch_error(promise.update, promise.exception)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                else:
+                    self.logger.exception(
+                        'No error handlers are set for async function, logging exception.',
+                        exc_info=promise.exception)
+            else:
+                self.update_persistence(update=promise.update)
 
-    def run_async(self, func, *args, **kwargs):
+    def run_async(self, func, *args, update=None, error_handler=None, **kwargs):
         """Queue a function (with given args/kwargs) to be run asynchronously.
 
         Warning:
-            If you're using @run_async you cannot rely on adding custom attributes to
-            :class:`telegram.ext.CallbackContext`. See its docs for more info.
+            * If you're using ``@run_async``/:meth:`run_async` you cannot rely on adding custom
+              attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+            * When using :meth:`run_async` within an asynchronously run error handler (see
+              :meth:`add_error_handler`) and ``update`` is passed, you *must* also pass
+              ``error_handler``, as an exception within ``func`` would cause an infinite loop
+              otherwise.
+            * passing ``error_handler`` is not supported for the old style API, i.e.
+              ``use_context=False``
 
         Args:
             func (:obj:`callable`): The function to run in the thread.
-            *args (:obj:`tuple`, optional): Arguments to `func`.
-            **kwargs (:obj:`dict`, optional): Keyword arguments to `func`.
+            *args (:obj:`tuple`, optional): Arguments to ``func``.
+            update (:class:`telegram.Update`, optional): The update associated with the functions
+                call.
+            error_handler (:obj:`callable`, optional): An error handler to call in case of an
+                exception during the call of ``func``. The callback signature of ``error_handler``
+                must be
+
+                ``def error_handler(context: CallbackContext, *args, **kwargs)``
+
+                or
+
+                ``def error_handler(update: Update, context: CallbackContext, *args, **kwargs)``
+
+                if ``update`` is passed, where ``*args`` and ``**kwargs`` are the arguments for
+                ``func``.
+            **kwargs (:obj:`dict`, optional): Keyword arguments to ``func``.
 
         Returns:
             Promise
 
         """
-        # TODO: handle exception in async threads
-        #       set a threading.Event to notify caller thread
-        promise = Promise(func, args, kwargs)
+        if self.use_context is False and error_handler:
path: telegram/ext/dispatcher.py
text: @user1:
```suggestion
        if error_handler and not self.use_context:
```

diff_hunk: @@ -229,30 +236,84 @@ def _pooled(self):
                 break
 
             promise.run()
+
             if isinstance(promise.exception, DispatcherHandlerStop):
                 self.logger.warning(
                     'DispatcherHandlerStop is not supported with async functions; func: %s',
                     promise.pooled_function.__name__)
+            elif isinstance(promise.exception, Exception):
+                if promise.pooled_function in self.error_handlers:
+                    self.logger.error('An error was raised while processing the update '
+                                      'and an uncaught error was raised while handling '
+                                      'the error with an error_handler')
+                elif promise.error_handler:
+                    try:
+                        context = CallbackContext.from_error(promise.update, promise.exception,
+                                                             self)
+                        if promise.update:
+                            promise.error_handler(promise.update, context, *promise.args,
+                                                  **promise.kwargs)
+                        else:
+                            promise.error_handler(context, *promise.args, **promise.kwargs)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                elif promise.update:
+                    try:
+                        self.dispatch_error(promise.update, promise.exception)
+                    except Exception:
+                        self.logger.exception('An error was raised while processing the update '
+                                              'and an uncaught error was raised while handling '
+                                              'the error with an error_handler')
+                else:
+                    self.logger.exception(
+                        'No error handlers are set for async function, logging exception.',
+                        exc_info=promise.exception)
+            else:
+                self.update_persistence(update=promise.update)
 
-    def run_async(self, func, *args, **kwargs):
+    def run_async(self, func, *args, update=None, error_handler=None, **kwargs):
         """Queue a function (with given args/kwargs) to be run asynchronously.
 
         Warning:
-            If you're using @run_async you cannot rely on adding custom attributes to
-            :class:`telegram.ext.CallbackContext`. See its docs for more info.
+            * If you're using ``@run_async``/:meth:`run_async` you cannot rely on adding custom
+              attributes to :class:`telegram.ext.CallbackContext`. See its docs for more info.
+            * When using :meth:`run_async` within an asynchronously run error handler (see
+              :meth:`add_error_handler`) and ``update`` is passed, you *must* also pass
+              ``error_handler``, as an exception within ``func`` would cause an infinite loop
+              otherwise.
+            * passing ``error_handler`` is not supported for the old style API, i.e.
+              ``use_context=False``
 
         Args:
             func (:obj:`callable`): The function to run in the thread.
-            *args (:obj:`tuple`, optional): Arguments to `func`.
-            **kwargs (:obj:`dict`, optional): Keyword arguments to `func`.
+            *args (:obj:`tuple`, optional): Arguments to ``func``.
+            update (:class:`telegram.Update`, optional): The update associated with the functions
+                call.
+            error_handler (:obj:`callable`, optional): An error handler to call in case of an
+                exception during the call of ``func``. The callback signature of ``error_handler``
+                must be
+
+                ``def error_handler(context: CallbackContext, *args, **kwargs)``
+
+                or
+
+                ``def error_handler(update: Update, context: CallbackContext, *args, **kwargs)``
+
+                if ``update`` is passed, where ``*args`` and ``**kwargs`` are the arguments for
+                ``func``.
+            **kwargs (:obj:`dict`, optional): Keyword arguments to ``func``.
 
         Returns:
             Promise
 
         """
-        # TODO: handle exception in async threads
-        #       set a threading.Event to notify caller thread
-        promise = Promise(func, args, kwargs)
+        if self.use_context is False and error_handler:
+            raise ValueError('Passing `error_handler` is not supported for non context aware '
+                             'dispatchers!')
path: telegram/ext/dispatcher.py
text: @user1:
```suggestion
            raise ValueError('`error_handler` is not supported with a non context aware dispatcher')
```

diff_hunk: @@ -448,57 +512,58 @@ def update_persistence(self, update=None):
             update (:class:`telegram.Update`, optional): The update to process. If passed, only the
             corresponding ``user_data`` and ``chat_data`` will be updated.
         """
-        if self.persistence:
-            chat_ids = self.chat_data.keys()
-            user_ids = self.user_data.keys()
-
-            if isinstance(update, Update):
-                if update.effective_chat:
-                    chat_ids = [update.effective_chat.id]
-                else:
-                    chat_ids = []
-                if update.effective_user:
-                    user_ids = [update.effective_user.id]
-                else:
-                    user_ids = []
-
-            if self.persistence.store_bot_data:
-                try:
-                    self.persistence.update_bot_data(self.bot_data)
-                except Exception as e:
-                    try:
-                        self.dispatch_error(update, e)
-                    except Exception:
-                        message = 'Saving bot data raised an error and an ' \
-                                  'uncaught error was raised while handling ' \
-                                  'the error with an error_handler'
-                        self.logger.exception(message)
-            if self.persistence.store_chat_data:
-                for chat_id in chat_ids:
+        with self._update_persistence_lock:
path: telegram/ext/dispatcher.py
text: @user1:
Try to keep a minimal diff (aka be source control friendly)

Please change to something like this:

```python
def update_persistence(self, update=None):
    with self._update_persistence_lock:
        self.__update_persistence(update)

def __update_persistence(self, update):
    # Assumes we're behind a lock
    .... all old code ...
```



@author:
Great tip! Thanks for that, I adopted it :)

diff_hunk: @@ -141,6 +146,7 @@ def __init__(self,
         self.user_data = defaultdict(dict)
         self.chat_data = defaultdict(dict)
         self.bot_data = {}
+        self._update_persistence_lock = Lock()
         if persistence:
             if not isinstance(persistence, BasePersistence):
                 raise TypeError("persistence should be based on telegram.ext.BasePersistence")
path: telegram/ext/dispatcher.py
text: @user1:
```suggestion
                raise TypeError("persistence must be based on telegram.ext.BasePersistence")
```

diff_hunk: @@ -229,30 +236,84 @@ def _pooled(self):
                 break
 
             promise.run()
+
             if isinstance(promise.exception, DispatcherHandlerStop):
                 self.logger.warning(
                     'DispatcherHandlerStop is not supported with async functions; func: %s',
                     promise.pooled_function.__name__)
+            elif isinstance(promise.exception, Exception):
+                if promise.pooled_function in self.error_handlers:
+                    self.logger.error('An error was raised while processing the update '
+                                      'and an uncaught error was raised while handling '
+                                      'the error with an error_handler')
path: telegram/ext/dispatcher.py
text: @user1:
Consider adding explicit `return` here. To make sure that this is the last thing executed.
The problem is that the function is very long and it can make the naive reader / developer add things afterwards missing the whole point of "stop it now!".

Also I would prefer a shorter sentence explaining the problem.

```suggestion
                    self.logger.error('An uncaught error was raised by an error_handler')
                    return
```

@author:
`return` statement: good idea. wording: I just used the wording that we use everywhere, i.e. also in `process_update` and in `JobQueue` … Should I change everywhere?

diff_hunk: @@ -45,6 +45,8 @@ class CallbackQueryHandler(Handler):
             the callback function.
         pass_chat_data (:obj:`bool`): Determines whether ``chat_data`` will be passed to
             the callback function.
+        run_async (:obj:`bool`): Determines whether this handlers callback function will be run
+            asynchronously.
path: telegram/ext/callbackqueryhandler.py
text: @user1:
```suggestion
        run_async (:obj:`bool`): Determines whether the callback will run asynchronously.
```