mirror of https://github.com/zulip/zulip.git
234 lines
7.1 KiB
Python
234 lines
7.1 KiB
Python
from enum import Enum
|
|
from typing import Any, Dict, List, Type, TypeVar, Optional
|
|
from typing_extensions import NoReturn
|
|
|
|
from django.utils.translation import ugettext as _
|
|
|
|
|
|
T = TypeVar("T", bound="AbstractEnum")
|
|
|
|
class AbstractEnum(Enum):
|
|
'''An enumeration whose members are used strictly for their names.'''
|
|
|
|
def __new__(cls: Type[T]) -> T:
|
|
obj = object.__new__(cls)
|
|
obj._value_ = len(cls.__members__) + 1
|
|
return obj
|
|
|
|
# Override all the `Enum` methods that use `_value_`.
|
|
|
|
def __repr__(self) -> str:
|
|
return str(self) # nocoverage
|
|
|
|
def value(self) -> None:
|
|
raise AssertionError("Not implemented")
|
|
|
|
def __reduce_ex__(self, proto: object) -> NoReturn:
|
|
raise AssertionError("Not implemented")
|
|
|
|
class ErrorCode(AbstractEnum):
|
|
BAD_REQUEST = () # Generic name, from the name of HTTP 400.
|
|
REQUEST_VARIABLE_MISSING = ()
|
|
REQUEST_VARIABLE_INVALID = ()
|
|
INVALID_JSON = ()
|
|
BAD_IMAGE = ()
|
|
REALM_UPLOAD_QUOTA = ()
|
|
BAD_NARROW = ()
|
|
CANNOT_DEACTIVATE_LAST_USER = ()
|
|
MISSING_HTTP_EVENT_HEADER = ()
|
|
STREAM_DOES_NOT_EXIST = ()
|
|
UNAUTHORIZED_PRINCIPAL = ()
|
|
UNEXPECTED_WEBHOOK_EVENT_TYPE = ()
|
|
BAD_EVENT_QUEUE_ID = ()
|
|
CSRF_FAILED = ()
|
|
INVITATION_FAILED = ()
|
|
INVALID_ZULIP_SERVER = ()
|
|
INVALID_MARKDOWN_INCLUDE_STATEMENT = ()
|
|
REQUEST_CONFUSING_VAR = ()
|
|
INVALID_API_KEY = ()
|
|
|
|
class JsonableError(Exception):
|
|
'''A standardized error format we can turn into a nice JSON HTTP response.
|
|
|
|
This class can be invoked in a couple ways.
|
|
|
|
* Easiest, but completely machine-unreadable:
|
|
|
|
raise JsonableError(_("No such widget: {}").format(widget_name))
|
|
|
|
The message may be passed through to clients and shown to a user,
|
|
so translation is required. Because the text will vary depending
|
|
on the user's language, it's not possible for code to distinguish
|
|
this error from others in a non-buggy way.
|
|
|
|
* Fully machine-readable, with an error code and structured data:
|
|
|
|
class NoSuchWidgetError(JsonableError):
|
|
code = ErrorCode.NO_SUCH_WIDGET
|
|
data_fields = ['widget_name']
|
|
|
|
def __init__(self, widget_name: str) -> None:
|
|
self.widget_name: str = widget_name
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("No such widget: {widget_name}")
|
|
|
|
raise NoSuchWidgetError(widget_name)
|
|
|
|
Now both server and client code see a `widget_name` attribute
|
|
and an error code.
|
|
|
|
Subclasses may also override `http_status_code`.
|
|
'''
|
|
|
|
# Override this in subclasses, as needed.
|
|
code: ErrorCode = ErrorCode.BAD_REQUEST
|
|
|
|
# Override this in subclasses if providing structured data.
|
|
data_fields: List[str] = []
|
|
|
|
# Optionally override this in subclasses to return a different HTTP status,
|
|
# like 403 or 404.
|
|
http_status_code: int = 400
|
|
|
|
def __init__(self, msg: str) -> None:
|
|
# `_msg` is an implementation detail of `JsonableError` itself.
|
|
self._msg: str = msg
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
'''Override in subclasses. Gets the items in `data_fields` as format args.
|
|
|
|
This should return (a translation of) a string literal.
|
|
The reason it's not simply a class attribute is to allow
|
|
translation to work.
|
|
'''
|
|
# Secretly this gets one more format arg not in `data_fields`: `_msg`.
|
|
# That's for the sake of the `JsonableError` base logic itself, for
|
|
# the simplest form of use where we just get a plain message string
|
|
# at construction time.
|
|
return '{_msg}'
|
|
|
|
#
|
|
# Infrastructure -- not intended to be overridden in subclasses.
|
|
#
|
|
|
|
@property
|
|
def msg(self) -> str:
|
|
format_data = dict(((f, getattr(self, f)) for f in self.data_fields),
|
|
_msg=getattr(self, '_msg', None))
|
|
return self.msg_format().format(**format_data)
|
|
|
|
@property
|
|
def data(self) -> Dict[str, Any]:
|
|
return dict(((f, getattr(self, f)) for f in self.data_fields),
|
|
code=self.code.name)
|
|
|
|
def to_json(self) -> Dict[str, Any]:
|
|
d = {'result': 'error', 'msg': self.msg}
|
|
d.update(self.data)
|
|
return d
|
|
|
|
def __str__(self) -> str:
|
|
return self.msg
|
|
|
|
class StreamDoesNotExistError(JsonableError):
|
|
code = ErrorCode.STREAM_DOES_NOT_EXIST
|
|
data_fields = ['stream']
|
|
|
|
def __init__(self, stream: str) -> None:
|
|
self.stream = stream
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("Stream '{stream}' does not exist")
|
|
|
|
class StreamWithIDDoesNotExistError(JsonableError):
|
|
code = ErrorCode.STREAM_DOES_NOT_EXIST
|
|
data_fields = ['stream_id']
|
|
|
|
def __init__(self, stream_id: int) -> None:
|
|
self.stream_id = stream_id
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("Stream with ID '{stream_id}' does not exist")
|
|
|
|
class CannotDeactivateLastUserError(JsonableError):
|
|
code = ErrorCode.CANNOT_DEACTIVATE_LAST_USER
|
|
data_fields = ['is_last_admin', 'entity']
|
|
|
|
def __init__(self, is_last_admin: bool) -> None:
|
|
self.is_last_admin = is_last_admin
|
|
self.entity = _("organization administrator") if is_last_admin else _("user")
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("Cannot deactivate the only {entity}.")
|
|
|
|
class InvalidMarkdownIncludeStatement(JsonableError):
|
|
code = ErrorCode.INVALID_MARKDOWN_INCLUDE_STATEMENT
|
|
data_fields = ['include_statement']
|
|
|
|
def __init__(self, include_statement: str) -> None:
|
|
self.include_statement = include_statement
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("Invalid markdown include statement: {include_statement}")
|
|
|
|
class RateLimited(Exception):
|
|
def __init__(self, msg: str="") -> None:
|
|
super().__init__(msg)
|
|
|
|
class InvalidJSONError(JsonableError):
|
|
code = ErrorCode.INVALID_JSON
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("Malformed JSON")
|
|
|
|
class OrganizationAdministratorRequired(JsonableError):
|
|
code: ErrorCode = ErrorCode.UNAUTHORIZED_PRINCIPAL
|
|
|
|
ADMIN_REQUIRED_MESSAGE = _("Must be an organization administrator")
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(self.ADMIN_REQUIRED_MESSAGE)
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return OrganizationAdministratorRequired.ADMIN_REQUIRED_MESSAGE
|
|
|
|
class BugdownRenderingException(Exception):
|
|
pass
|
|
|
|
class InvalidAPIKeyError(JsonableError):
|
|
code = ErrorCode.INVALID_API_KEY
|
|
http_status_code = 401
|
|
|
|
def __init__(self) -> None:
|
|
pass
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("Invalid API key")
|
|
|
|
class InvalidAPIKeyFormatError(InvalidAPIKeyError):
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("Malformed API key")
|
|
|
|
class UnexpectedWebhookEventType(JsonableError):
|
|
code = ErrorCode.UNEXPECTED_WEBHOOK_EVENT_TYPE
|
|
data_fields = ['webhook_name', 'event_type']
|
|
|
|
def __init__(self, webhook_name: str, event_type: Optional[str]) -> None:
|
|
self.webhook_name = webhook_name
|
|
self.event_type = event_type
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return _("The '{event_type}' event isn't currently supported by the {webhook_name} webhook")
|