"""Unit tests"""
-class TweepyErrorTests(unittest.TestCase):
-
- def testpickle(self):
- """Verify exceptions can be pickled and unpickled."""
- import pickle
- from tweepy.error import TweepError
-
- e = TweepError('no reason', {'status': 200})
- e2 = pickle.loads(pickle.dumps(e))
-
- self.assertEqual(e.reason, e2.reason)
- self.assertEqual(e.response, e2.response)
-
-
class TweepyAPITests(TweepyTestCase):
#@tape.use_cassette('testfailure.json')
from .config import create_auth
from tweepy import API
-from tweepy.error import TweepError
+from tweepy.errors import HTTPException
testratelimit = 'TEST_RATE_LIMIT' in os.environ
for user_id in test_user_ids:
try:
self.api.user_timeline(user_id=user_id, count=1, include_rts=True)
- except TweepError as e:
- # continue if we're not autherized to access the user's timeline or she doesn't exist anymore
- if e.response is not None and e.response.status in {401, 404}:
+ except HTTPException as e:
+ # continue if we're not autherized to access the user's timeline or user doesn't exist anymore
+ if e.response.status_code in (401, 404):
continue
raise e
from tweepy.auth import AppAuthHandler, OAuthHandler
from tweepy.cache import Cache, FileCache, MemoryCache
from tweepy.cursor import Cursor
-from tweepy.error import RateLimitError, TweepError
-from tweepy.errors import TweepyException
+from tweepy.errors import HTTPException, TooManyRequests, TweepyException
from tweepy.models import DirectMessage, Friendship, ModelFactory, SavedSearch, SearchResults, Status, User
from tweepy.streaming import Stream
import requests
-from tweepy.error import is_rate_limit_error_message, RateLimitError, TweepError
-from tweepy.errors import TweepyException
+from tweepy.errors import HTTPException, TooManyRequests, TweepyException
from tweepy.models import Model
from tweepy.parsers import ModelParser, Parser
from tweepy.utils import list_to_csv
# If an error was returned, throw an exception
self.last_response = resp
+ if resp.status_code == 429:
+ raise TooManyRequests(resp)
if resp.status_code and not 200 <= resp.status_code < 300:
- try:
- error_msg, api_error_code = parser.parse_error(resp.text)
- except Exception:
- error_msg = f"Twitter error response: status code = {resp.status_code}"
- api_error_code = None
-
- if is_rate_limit_error_message(error_msg):
- raise RateLimitError(error_msg, resp)
- else:
- raise TweepError(error_msg, resp, api_code=api_error_code)
+ raise HTTPException(resp)
# Parse the response payload
return_cursors = return_cursors or 'cursor' in params or 'next' in params
+++ /dev/null
-# Tweepy
-# Copyright 2009-2021 Joshua Roesslein
-# See LICENSE for details.
-
-
-class TweepError(Exception):
- """Tweepy exception"""
-
- def __init__(self, reason, response=None, api_code=None):
- self.reason = str(reason)
- self.response = response
- self.api_code = api_code
- super().__init__(reason)
-
- def __str__(self):
- return self.reason
-
-
-def is_rate_limit_error_message(message):
- """Check if the supplied error message belongs to a rate limit error."""
- return (isinstance(message, list) and len(message) > 0 and
- 'code' in message[0] and message[0]['code'] == 88)
-
-
-class RateLimitError(TweepError):
- """Exception for Tweepy hitting the rate limit."""
- # RateLimitError has the exact same properties and inner workings
- # as TweepError for backwards compatibility reasons.
- pass
# Copyright 2009-2021 Joshua Roesslein
# See LICENSE for details.
+import json
+
class TweepyException(Exception):
"""Base exception for Tweepy"""
pass
+
+
+class HTTPException(TweepyException):
+ """Exception raised when an HTTP request fails"""
+
+ def __init__(self, response):
+ self.response = response
+
+ self.api_errors = []
+ self.api_codes = []
+ self.api_messages = []
+
+ try:
+ response_json = response.json()
+ except json.JSONDecodeError:
+ super().__init__(f"{response.status_code} {response.reason}")
+ else:
+ error_text = []
+ for error in response_json.get("errors"):
+ self.api_errors.append(error)
+ # Use := when support for Python 3.7 is dropped
+ if "code" in error:
+ self.api_codes.append(error["code"])
+ if "message" in error:
+ self.api_messages.append(error["message"])
+ if "code" in error and "message" in error:
+ error_text.append(f"{error['code']} - {error['message']}")
+ elif "message" in error:
+ error_text.append(error["message"])
+ error_text = '\n'.join(error_text)
+ super().__init__(
+ f"{response.status_code} {response.reason}\n{error_text}"
+ )
+
+
+class TooManyRequests(HTTPException):
+ """Exception raised for a 429 HTTP status code"""
+ pass
"""
raise NotImplementedError
- def parse_error(self, payload):
- """
- Parse the error message and api error code from payload.
- Return them as an (error_msg, error_code) tuple. If unable to parse the
- message, throw an exception and default error message will be used.
- """
- raise NotImplementedError
-
class RawParser(Parser):
def parse(self, payload, *args, **kwargs):
return payload
- def parse_error(self, payload):
- return payload
-
class JSONParser(Parser):
return json, json['next_cursor']
return json
- def parse_error(self, payload):
- error_object = json_lib.loads(payload)
-
- if 'error' in error_object:
- reason = error_object['error']
- api_code = error_object.get('code')
- else:
- reason = error_object['errors']
- api_code = [error.get('code') for error in
- reason if error.get('code')]
- api_code = api_code[0] if len(api_code) == 1 else api_code
-
- return reason, api_code
-
class ModelParser(JSONParser):
projects / tweepy.git / commitdiff