from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, Dict, Mapping, Optional, Sequence, Union
from uuid import uuid4
from slack.python_compatibility import format_exception_only
from slack.shared import shared
if TYPE_CHECKING:
from slack_api.slack_common import SlackErrorResponse
from slack_rtm.slack_rtm_message import SlackRtmMessage
from slack.slack_workspace import SlackWorkspace
class HttpError(Exception):
def __init__(
self,
url: str,
options: Dict[str, str],
return_code: Optional[int],
http_status_code: Optional[int],
error: str,
):
super().__init__(
f"{self.__class__.__name__}: url='{url}', return_code={return_code}, http_status_code={http_status_code}, error='{error}'"
)
self.url = url
self.options = options
self.return_code = return_code
self.http_status_code = http_status_code
self.error = error
class SlackApiError(Exception):
def __init__(
self,
workspace: SlackWorkspace,
method: str,
response: SlackErrorResponse,
params: Mapping[
str, Union[str, int, bool, Sequence[str], Sequence[int], Sequence[bool]]
] = {},
):
super().__init__(
f"{self.__class__.__name__}: workspace={workspace}, method='{method}', params={params}, response={response}"
)
self.workspace = workspace
self.method = method
self.params = params
self.response = response
class SlackRtmError(Exception):
def __init__(
self,
workspace: SlackWorkspace,
exception: BaseException,
message_json: SlackRtmMessage,
):
super().__init__(
f"{self.__class__.__name__}: workspace={workspace}, exception=`{format_exception_only_str(exception)}`"
)
super().with_traceback(exception.__traceback__)
self.workspace = workspace
self.exception = exception
self.message_json = message_json
class SlackError(Exception):
def __init__(
self, workspace: SlackWorkspace, error: str, data: Optional[object] = None
):
super().__init__(
f"{self.__class__.__name__}: workspace={workspace}, error={error}"
)
self.workspace = workspace
self.error = error
self.data = data
@dataclass
class UncaughtError:
id: str = field(init=False)
exception: BaseException
def __post_init__(self):
self.id = str(uuid4())
self.time = datetime.now()
def format_exception_only_str(exc: BaseException) -> str:
return format_exception_only(exc)[-1].strip()
def store_uncaught_error(uncaught_error: UncaughtError) -> None:
shared.uncaught_errors.append(uncaught_error)
def store_and_format_uncaught_error(uncaught_error: UncaughtError) -> str:
store_uncaught_error(uncaught_error)
e = uncaught_error.exception
stack_msg_command = f"/slack debug error {uncaught_error.id}"
stack_msg = f"run `{stack_msg_command}` for the stack trace"
if isinstance(e, HttpError):
return (
f"Error calling URL {e.url}: return code: {e.return_code}, "
f"http status code: {e.http_status_code}, error: {e.error} ({stack_msg})"
)
elif isinstance(e, SlackApiError):
return (
f"Error from Slack API method {e.method} with params {e.params} for workspace "
f"{e.workspace.name}: {e.response} ({stack_msg})"
)
elif isinstance(e, SlackRtmError):
return (
f"Error while handling Slack event of type '{e.message_json['type']}' for workspace "
f"{e.workspace.name}: {format_exception_only_str(e.exception)} ({stack_msg}, "
f"run `{stack_msg_command} -data` for the event data)"
)
elif isinstance(e, SlackError):
return (
f"Error occurred in workspace {e.workspace.name}: {e.error} ({stack_msg})"
)
else:
return f"Unknown error occurred: {format_exception_only_str(e)} ({stack_msg})"
def store_and_format_exception(e: BaseException) -> str:
uncaught_error = UncaughtError(e)
return store_and_format_uncaught_error(uncaught_error)
