add some exceptions handling

Author: ape364

exchanges/base.py

@@ -4,9 +4,10 @@
 from enum import Enum
 from logging import getLogger
 
+import aiohttp
 from aiohttp import ClientSession
 
-from exchanges.exceptions import WrongContentTypeException
+from exchanges.exceptions import WrongContentTypeException, BaseExchangeException, InvalidResponseException
 
 Order = namedtuple('Order', 'exchange_id order_id type pair price amount state')
 
@@ -45,31 +46,32 @@ def __init__(self, key, secret):
     def check_keys(cls, api: str, secret: str) -> bool:
         return cls.api_regex.match(api) and cls.secret_regex.match(secret)
 
-    @staticmethod
-    async def post(url: str, headers: dict = None, data: dict = None) -> dict:
-        async with ClientSession() as s:
-            resp = await s.post(url, data=data, headers=headers)
-            return await resp.json()
-
-    async def get(self, url: str, headers: dict = None) -> dict:
+    async def request(self, url, headers, method='get', data=None):
         attempt, delay = 1, 1
         async with ClientSession() as s:
+            session_method = s.__getattribute__(method.lower())
             while True:
                 try:
-                    resp = await s.get(url, headers=headers)
+                    resp = await session_method(url=url, headers=headers, data=data)
                     if resp.content_type != 'application/json':
-                        raise WrongContentTypeException(
-                            f'Unexpected content type {resp.content_type!r}. URL: {url}, headers: {headers}'
-                        )
-                except WrongContentTypeException as e:
-                    getLogger().error(f'attempt {attempt}/{self.attempts_limit}, next attempt in {delay} seconds')
+                        raise WrongContentTypeException(f'Unexpected content type {resp.content_type!r} at URL {url}.')
+                    json_resp = await resp.json()
+                    self._raise_if_error(json_resp)
+                    return json_resp
+                except (aiohttp.client_exceptions.ClientResponseError, BaseExchangeException) as e:
+                    getLogger().error(f'attempt {attempt}/{self.attempts_limit}, next in {delay} seconds...')
                     getLogger().exception(e)
                     attempt += 1
                     if attempt > self.attempts_limit:
-                        return {}
+                        raise InvalidResponseException(e)
                     await asyncio.sleep(delay)
                     delay *= 2
-                return await resp.json()
+
+    async def post(self, url: str, headers: dict = None, data: dict = None) -> dict:
+        return await self.request(url, headers, 'post', data)
+
+    async def get(self, url: str, headers: dict = None) -> dict:
+        return await self.request(url, headers)
 
     @abstractmethod
     async def order_history(self) -> [str, ]:
@@ -96,3 +98,8 @@ def _get_ticker_url(self, pair):
     def _order_state(order: dict) -> State:
         '''Returns state of the api order.'''
 
+    @abstractmethod
+    def _raise_if_error(self, response: dict) -> bool:
+        '''Raises BaseExchangeException if there is an errors in API response.
+        :raises BaseExchangeException:
+        '''

exchanges/bittrex.py

@@ -36,8 +36,6 @@ async def order_info(self, order_id: str) -> dict:
         params = {'uuid': order_id}
         headers, url = self.get_headers_url(method_url, **params)
         resp = await self.get(url, headers)
-        if not resp['success']:
-            raise BittrexApiException(resp['message'])
         order = resp['result']
         return Order(
             self.api_id,
@@ -75,3 +73,7 @@ def get_headers_url(self, method_url, **params):
         })
         url = f'{method_url}?{urlencode(params)}'
         return {'apisign': hmac.new(self._secret.encode(), url.encode(), hashlib.sha512).hexdigest()}, url
+
+    def _raise_if_error(self, response: dict):
+        if not response['success']:
+            raise BittrexApiException(response['message'])

exchanges/exceptions.py

@@ -2,5 +2,9 @@ class BaseExchangeException(BaseException):
     pass
 
 
+class InvalidResponseException(BaseExchangeException):
+    pass
+
+
 class WrongContentTypeException(BaseExchangeException):
     pass

exchanges/kraken.py

@@ -54,8 +54,6 @@ async def order_history(self) -> [str, ]:
         }
 
         resp = await self.post(self.BASE_URL + method_url, headers, data)
-        if resp['error']:
-            raise KrakenApiException('\n'.join(resp['error']))
 
         return {order_id for order_id in resp['result']['closed']}
 
@@ -73,8 +71,6 @@ async def order_info(self, order_id: str) -> Order:
         }
 
         resp = await self.post(self.BASE_URL + method_url, headers, data)
-        if resp['error']:
-            raise KrakenApiException('\n'.join(resp['error']))
 
         order = resp['result'][order_id]
         descr = order['descr']
@@ -121,3 +117,7 @@ async def parse_pair(self, pair):
             raise AmbiguousResultException(f'More than 1 result to pair {pair}: {pairs}')
         _, pair_info = result.popitem()
         return f"{pair_info['base']}-{pair_info['quote']}"
+
+    def _raise_if_error(self, response: dict) -> bool:
+        if response['error']:
+            raise KrakenApiException('\n'.join(response['error']))

exchanges/liqui.py

@@ -1,13 +1,9 @@
-import asyncio
 import hashlib
 import hmac
 import re
-from logging import getLogger
 from time import time
 from urllib.parse import urlencode
 
-import aiohttp
-
 from exchanges.base import BaseApi, Order, State
 from exchanges.exceptions import BaseExchangeException
 
@@ -52,30 +48,21 @@ def _get_ticker_url(self, pair):
         return f'https://liqui.io/#/exchange/{cur_from}_{cur_to}'
 
     async def _tapi(self, **params):
-        attempt, delay = 1, 1
-        while True:
-            try:
-                params['nonce'] = int(time())
-                data = await self.post(
-                    'https://api.liqui.io/tapi',
-                    headers={'Key': self._key, 'Sign': self._sign(params)},
-                    data=params
-                )
-                if 'error' in data:
-                    if data['error'] == 'no orders':
-                        raise NoOrdersException(data['error'])
-                    raise LiquiApiException(data['error'])
-                return data.get('return', data)
-            except (LiquiApiException, aiohttp.client_exceptions.ClientResponseError) as e:
-                getLogger().error(f'attempt {attempt}/{self.attempts_limit}, next attempt in {delay} seconds')
-                getLogger().exception(e)
-                attempt += 1
-                if attempt > self.attempts_limit:
-                    return {}
-                await asyncio.sleep(delay)
-                delay *= 2
+        params['nonce'] = int(time())
+        resp = await self.post(
+            'https://api.liqui.io/tapi',
+            headers={'Key': self._key, 'Sign': self._sign(params)},
+            data=params
+        )
+        return resp.get('return', resp)
 
     def _sign(self, data):
         if isinstance(data, dict):
             data = urlencode(data)
         return hmac.new(self._secret.encode(), data.encode(), hashlib.sha512).hexdigest()
+
+    def _raise_if_error(self, response: dict):
+        if 'error' in response:
+            if response['error'] == 'no orders':
+                raise NoOrdersException(response['error'])
+            raise LiquiApiException(response['error'])

order_checker.py

@@ -7,6 +7,7 @@
 import settings
 from exchanges import exchange_apis
 from exchanges.base import state_text
+from exchanges.exceptions import BaseExchangeException
 
 
 class OrderChecker:
@@ -24,7 +25,12 @@ async def check(self):
                 api = exchange_api(api_key, secret_key)
 
                 db_orders = await db.get_order_ids(exchange_id, uid)
-                api_orders = await api.order_history()
+                try:
+                    api_orders = await api.order_history()
+                except BaseExchangeException as e:
+                    getLogger().error(f'Error while parsing exchange {exchange_name!r} of user id {uid}, skipping...')
+                    getLogger().exception(e)
+                    continue
 
                 new_orders = api_orders - db_orders
 
@@ -36,7 +42,15 @@ async def check(self):
                 await db.add_orders((uid, exchange_id, order_id) for order_id in new_orders)
 
                 for order_id in new_orders:
-                    order = await api.order_info(order_id)
+                    try:
+                        order = await api.order_info(order_id)
+                    except BaseExchangeException as e:
+                        getLogger().error(
+                            f'Error while fetching order {order_id!r} of user id {uid} '
+                            f'at exchange {exchange_name!r}, skipping...'
+                        )
+                        getLogger().exception(e)
+                        continue
                     state = state_text[order.state]
                     getLogger().info(f'Order {order_id} of user {uid} at exchange {exchange_name!r} '
                                      f'with id {exchange_id} is {state}.')

settings.py

@@ -6,3 +6,5 @@
 DATABASE_URL = os.environ['DATABASE_URL']
 
 CHECK_INTERVAL = int(os.environ['NOTIFY_BOT_CHECK_INTERVAL'])  # seconds
+
+REQUEST_ATTEMPTS_LIMIT = int(os.environ['NOTIFY_BOT_ATTEMPTS_LIMIT'])