Skip to content

refactor: sourcery-AI-refactor #621

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Mar 13, 2022
Merged

refactor: sourcery-AI-refactor #621

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 80 additions & 89 deletions interactions/api/gateway.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -280,7 +280,7 @@ async def wait_until_ready(self):
"""Waits for the client to become ready according to the Gateway."""
await self.ready.wait()

def _dispatch_event(self, event: str, data: dict) -> None:
def _dispatch_event(self, event: str, data: dict) -> None: # sourcery no-metrics
"""
Dispatches an event from the Gateway.

Expand All @@ -292,96 +292,87 @@ def _dispatch_event(self, event: str, data: dict) -> None:
path: str = "interactions"
path += ".models" if event == "INTERACTION_CREATE" else ".api.models"

if event != "TYPING_START":
if event != "INTERACTION_CREATE":
name: str = event.lower()
try:
_event_path: list = [section.capitalize() for section in name.split("_")]
_name: str = (
_event_path[0] if len(_event_path) < 3 else "".join(_event_path[:-1])
)
__obj: object = getattr(__import__(path), _name)

if name in {"_create", "_add"}:
data["_client"] = self._http

self._dispatch.dispatch(f"on_{name}", __obj(**data)) # noqa
except AttributeError as error:
log.fatal(f"An error occured dispatching {name}: {error}")
if event == "INTERACTION_CREATE":
if not data.get("type"):
log.warning(
"Context is being created for the interaction, but no type is specified. Skipping..."
)
else:
if not data.get("type"):
log.warning(
"Context is being created for the interaction, but no type is specified. Skipping..."
)
else:
_context = self.__contextualize(data)
_name: str = ""
__args: list = [_context]
__kwargs: dict = {}

if data["type"] == InteractionType.APPLICATION_COMMAND:
_name = f"command_{_context.data.name}"

if _context.data._json.get("options"):
for option in _context.data.options:
_type = self.__option_type_context(
_context,
(
option["type"]
if isinstance(option, dict)
else option.type.value
),
)
if _type:
if isinstance(option, dict):
_type[option["value"]]._client = self._http
option.update({"value": _type[option["value"]]})
else:
_type[option.value]._client = self._http
option._json.update({"value": _type[option.value]})
_option = self.__sub_command_context(option, _context)
__kwargs.update(_option)

self._dispatch.dispatch("on_command", _context)
elif data["type"] == InteractionType.MESSAGE_COMPONENT:
_name = f"component_{_context.data.custom_id}"

if _context.data._json.get("values"):
__args.append(_context.data.values)

self._dispatch.dispatch("on_component", _context)
elif data["type"] == InteractionType.APPLICATION_COMMAND_AUTOCOMPLETE:
_name = f"autocomplete_{_context.data.id}"

if _context.data._json.get("options"):
for option in _context.data.options:
__name, _value = self.__sub_command_context(option, _context)
_name += f"_{__name}" if __name else ""

if _value:
__args.append(_value)

self._dispatch.dispatch("on_autocomplete", _context)
elif data["type"] == InteractionType.MODAL_SUBMIT:
_name = f"modal_{_context.data.custom_id}"

if _context.data._json.get("components"):
for component in _context.data.components:
if component.get("components"):
__args.append(
[_value["value"] for _value in component["components"]][0]
)
# sourcery skip: extract-method
_context = self.__contextualize(data)
_name: str = ""
__args: list = [_context]
__kwargs: dict = {}

if data["type"] == InteractionType.APPLICATION_COMMAND:
_name = f"command_{_context.data.name}"

if _context.data._json.get("options"):
for option in _context.data.options:
_type = self.__option_type_context(
_context,
(option["type"] if isinstance(option, dict) else option.type.value),
)
if _type:
if isinstance(option, dict):
_type[option["value"]]._client = self._http
option.update({"value": _type[option["value"]]})
else:
__args.append(
[_value.value for _value in component.components][0]
)

self._dispatch.dispatch("on_modal", _context)

self._dispatch.dispatch(_name, *__args, **__kwargs)
self._dispatch.dispatch("on_interaction", _context)
self._dispatch.dispatch("on_interaction_create", _context)

_type[option.value]._client = self._http
option._json.update({"value": _type[option.value]})
_option = self.__sub_command_context(option, _context)
__kwargs.update(_option)

self._dispatch.dispatch("on_command", _context)
elif data["type"] == InteractionType.MESSAGE_COMPONENT:
_name = f"component_{_context.data.custom_id}"

if _context.data._json.get("values"):
__args.append(_context.data.values)

self._dispatch.dispatch("on_component", _context)
elif data["type"] == InteractionType.APPLICATION_COMMAND_AUTOCOMPLETE:
_name = f"autocomplete_{_context.data.id}"

if _context.data._json.get("options"):
for option in _context.data.options:
__name, _value = self.__sub_command_context(option, _context)
_name += f"_{__name}" if __name else ""

if _value:
__args.append(_value)

self._dispatch.dispatch("on_autocomplete", _context)
elif data["type"] == InteractionType.MODAL_SUBMIT:
_name = f"modal_{_context.data.custom_id}"

if _context.data._json.get("components"):
for component in _context.data.components:
if component.get("components"):
__args.append(
[_value["value"] for _value in component["components"]][0]
)
else:
__args.append([_value.value for _value in component.components][0])

self._dispatch.dispatch("on_modal", _context)

self._dispatch.dispatch(_name, *__args, **__kwargs)
self._dispatch.dispatch("on_interaction", _context)
self._dispatch.dispatch("on_interaction_create", _context)
elif event != "TYPING_START":
name: str = event.lower()
try:
_event_path: list = [section.capitalize() for section in name.split("_")]
_name: str = _event_path[0] if len(_event_path) < 3 else "".join(_event_path[:-1])
__obj: object = getattr(__import__(path), _name)

if name in {"_create", "_add"}:
data["_client"] = self._http

self._dispatch.dispatch(f"on_{name}", __obj(**data)) # noqa
except AttributeError as error:
log.fatal(f"An error occured dispatching {name}: {error}")
self._dispatch.dispatch("raw_socket_create", data)

def __contextualize(self, data: dict) -> object:
Expand Down
16 changes: 8 additions & 8 deletions interactions/api/models/message.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -972,11 +972,11 @@ def set_field_at(
self.fields[index] = EmbedField(name=name, value=value, inline=inline)
self._json.update({"fields": [field._json for field in self.fields]})

except AttributeError:
raise AttributeError("No fields found in Embed")
except AttributeError as e:
raise AttributeError("No fields found in Embed") from e

except IndexError:
raise IndexError("No fields at this index")
except IndexError as e:
raise IndexError("No fields at this index") from e

def remove_field(self, index: int) -> None:
"""
Expand All @@ -990,11 +990,11 @@ def remove_field(self, index: int) -> None:
self.fields.pop(index)
self._json.update({"fields": [field._json for field in self.fields]})

except AttributeError:
raise AttributeError("No fields found in Embed")
except AttributeError as e:
raise AttributeError("No fields found in Embed") from e

except IndexError:
raise IndexError("Field not Found at index")
except IndexError as e:
raise IndexError("Field not Found at index") from e

def remove_author(self) -> None:
"""
Expand Down