# -*- coding: utf-8 -*-
from time import time
from . import data_processing
[docs]def get_error(data):
"""return the error if there is a corresponding exception"""
if isinstance(data, dict):
if "errors" in data:
error = data["errors"][0]
else:
error = data.get("error", None)
if isinstance(error, dict):
if error.get("code") in errors:
return error
[docs]async def throw(response, loads=None, encoding=None, **kwargs):
"""Get the response data if possible and raise an exception"""
if loads is None:
loads = data_processing.loads
data = await data_processing.read(response, loads=loads, encoding=encoding)
error = get_error(data)
if error is not None:
exception = errors[error["code"]]
raise exception(response=response, error=error, data=data, **kwargs)
if response.status in statuses:
exception = statuses[response.status]
raise exception(response=response, data=data, **kwargs)
# raise PeonyException if no specific exception was found
raise PeonyException(response=response, data=data, **kwargs)
[docs]class PeonyException(Exception):
"""Parent class of all the exceptions of Peony"""
def __init__(self, response=None, error=None, data=None, url=None, message=None):
"""
Add the response and data attributes
Extract message from the error if not explicitly given
"""
self.response = response
self.data = data
self.error = error
self.url = url
if not message:
message = self.get_message()
if url:
message += "\nurl: " + url
super().__init__(message)
[docs] def get_message(self):
if self.error is not None:
return self.error.get("message", self.error)
return str(self.data)
[docs]class PeonyUnavailableMethod(PeonyException):
def __init__(self, message):
super().__init__(message=message)
[docs]class PeonyDecodeError(PeonyException):
def __init__(self, exception, *args, **kwargs):
self.exception = exception
super().__init__(*args, **kwargs)
[docs] def get_message(self):
return "Could not decode response data:\n%s" % self.data
[docs]class StreamLimit(PeonyException):
pass
[docs]class ErrorDict(dict):
"""A dict to easily add exception associated to a code"""
[docs] def code(self, code):
"""Decorator to associate a code to an exception"""
def decorator(exception):
self[code] = exception
return exception
return decorator
statuses = ErrorDict()
errors = ErrorDict()
[docs]@statuses.code(304)
class HTTPNotModified(PeonyException):
pass
[docs]@statuses.code(400)
class HTTPBadRequest(PeonyException):
pass
[docs]@statuses.code(401)
class HTTPUnauthorized(PeonyException):
pass
[docs]@statuses.code(403)
class HTTPForbidden(PeonyException):
pass
[docs]@statuses.code(404)
class HTTPNotFound(PeonyException):
pass
[docs]@statuses.code(406)
class HTTPNotAcceptable(PeonyException):
pass
[docs]@statuses.code(409)
class HTTPConflict(PeonyException):
pass
[docs]@statuses.code(410)
class HTTPGone(PeonyException):
pass
[docs]@statuses.code(420)
class HTTPEnhanceYourCalm(PeonyException):
pass
[docs]@statuses.code(422)
class HTTPUnprocessableEntity(PeonyException):
pass
[docs]@statuses.code(429)
class HTTPTooManyRequests(PeonyException):
pass
[docs]@statuses.code(500)
class HTTPInternalServerError(PeonyException):
pass
[docs]@statuses.code(502)
class HTTPBadGateway(PeonyException):
pass
[docs]@statuses.code(503)
class HTTPServiceUnavailable(PeonyException):
pass
[docs]@statuses.code(504)
class HTTPGatewayTimeout(PeonyException):
pass
[docs]@errors.code(3)
class InvalidCoordinates(HTTPBadRequest):
pass
[docs]@errors.code(13)
class NoLocationAssociatedToIP(HTTPNotFound):
pass
[docs]@errors.code(17)
class NoUserMatchesQuery(HTTPNotFound):
pass
[docs]@errors.code(32)
class NotAuthenticated(HTTPUnauthorized):
pass
[docs]@errors.code(34)
class DoesNotExist(HTTPNotFound):
pass
[docs]@errors.code(36)
class CannotReportYourselfAsSpam(HTTPForbidden):
pass
[docs]@errors.code(38)
class ParameterMissing(HTTPForbidden):
pass
[docs]@errors.code(44)
class AttachmentURLInvalid(HTTPBadRequest):
pass
[docs]@errors.code(50)
class UserNotFound(HTTPNotFound):
pass
[docs]@errors.code(63)
class UserSuspended(HTTPNotFound):
pass
[docs]@errors.code(64)
class AccountSuspended(HTTPForbidden):
pass
[docs]@errors.code(68)
class MigrateToNewAPI(HTTPGone):
pass
[docs]@errors.code(87)
class ActionNotPermitted(HTTPForbidden):
pass
# TODO: check if that could be moved to RateLimitExceeded
[docs]@errors.code(88)
class RateLimitExceeded(HTTPTooManyRequests):
"""Exception raised on rate limit"""
@property
def reset(self):
"""
Time when the limit will be reset
Returns
-------
int
Time when the limit will be reset
"""
return int(self.response.headers.get("X-Rate-Limit-Reset", 0))
@property
def reset_in(self):
"""
Time in seconds until the limit will be reset
Returns
-------
int
Time in seconds until the limit will be reset
"""
return max(self.reset - time(), 0)
[docs]@errors.code(89)
class InvalidOrExpiredToken(HTTPForbidden):
pass
[docs]@errors.code(92)
class SSLRequired(HTTPForbidden):
pass
[docs]@errors.code(93)
class ApplicationNotAllowedToAccessDirectMessages(HTTPForbidden):
pass
[docs]@errors.code(99)
class UnableToVerifyCredentials(HTTPForbidden):
pass
[docs]@errors.code(130)
class OverCapacity(HTTPServiceUnavailable):
pass
[docs]@errors.code(131)
class InternalError(HTTPInternalServerError):
pass
[docs]@errors.code(136)
class Blocked(HTTPForbidden):
pass
[docs]@errors.code(135)
class CouldNotAuthenticate(HTTPUnauthorized):
pass
[docs]@errors.code(139)
class StatusAlreadyFavorited(HTTPForbidden):
pass
[docs]@errors.code(144)
class StatusNotFound(HTTPNotFound):
pass
[docs]@errors.code(150)
class CannotSendMessageToNonFollowers(HTTPForbidden):
pass
[docs]@errors.code(160)
class FollowRequestAlreadyChanged(HTTPForbidden):
pass
[docs]@errors.code(161)
class FollowLimit(HTTPForbidden):
pass
[docs]@errors.code(162)
class FollowBlocked(HTTPForbidden):
pass
[docs]@errors.code(185)
class StatusLimit(HTTPForbidden):
pass
[docs]@errors.code(187)
class DuplicatedStatus(HTTPForbidden):
pass
[docs]@errors.code(205)
class SpamReportLimit(HTTPForbidden):
pass
[docs]@errors.code(214)
class OwnerMustAllowDMFromAnyone(HTTPForbidden):
pass
[docs]@errors.code(215)
class BadAuthentication(HTTPBadRequest):
pass
[docs]@errors.code(220)
class AccessNotAllowedByCredentials(HTTPForbidden):
pass
[docs]@errors.code(226)
class AutomatedRequest(HTTPForbidden):
pass
[docs]@errors.code(251)
class RetiredEndpoint(HTTPGone):
pass
[docs]@errors.code(261)
class ReadOnlyApplication(HTTPForbidden):
pass
[docs]@errors.code(271)
class CannotMuteYourself(HTTPForbidden):
pass
[docs]@errors.code(272)
class NotMutingUser(HTTPForbidden):
pass
[docs]@errors.code(323)
class GIFNotAllowedWithMultipleImages(HTTPBadRequest):
pass
[docs]@errors.code(326)
class AccountLocked(HTTPForbidden):
pass
[docs]@errors.code(349)
class CannotSendMessageToUser(HTTPForbidden):
pass
[docs]@errors.code(354)
class DMCharacterLimit(HTTPForbidden):
pass
[docs]@errors.code(355)
class SubscriptionAlreadyExists(HTTPConflict):
pass
[docs]@errors.code(386)
class TooManyAttachmentTypes(HTTPForbidden):
pass
[docs]@errors.code(407)
class InvalidURL(HTTPBadRequest):
pass
[docs]@errors.code(415)
class CallbackURLNotApproved(HTTPForbidden):
pass
[docs]@errors.code(416)
class InvalidOrSuspendedApplication(HTTPUnauthorized):
pass
[docs]@errors.code(417)
class DesktopApplicationAuth(HTTPUnauthorized):
pass