aboutsummaryrefslogtreecommitdiffstats
path: root/slack/commands.py
diff options
context:
space:
mode:
authorTrygve Aaberge <trygveaa@gmail.com>2023-12-14 15:06:41 +0100
committerTrygve Aaberge <trygveaa@gmail.com>2024-02-18 11:32:54 +0100
commit57ad5138dc9bb5aa85ae4f58ee022f2ea2003b5d (patch)
treeabba0fe0258afb1fcbf88e3b7eb656a8ae677cb2 /slack/commands.py
parent71883e46b99a46e297a790fed59d27f1b85c92a5 (diff)
downloadwee-slack-57ad5138dc9bb5aa85ae4f58ee022f2ea2003b5d.tar.gz
Set options value to True if it's set without a value
It became very confusing to use None to represent it being set without a value, so use True instead. If it has a value it will always be a string, so you can still distinguish between an option without a value and an option with a truthy value.
Diffstat (limited to 'slack/commands.py')
-rw-r--r--slack/commands.py74
1 files changed, 27 insertions, 47 deletions
diff --git a/slack/commands.py b/slack/commands.py
index 9dc277d..f9d6919 100644
--- a/slack/commands.py
+++ b/slack/commands.py
@@ -7,7 +7,7 @@ import re
from dataclasses import dataclass
from functools import wraps
from itertools import chain
-from typing import Callable, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Callable, Dict, List, Optional, Tuple, Union
import weechat
@@ -24,6 +24,11 @@ from slack.task import run_async, sleep
from slack.util import get_callback_name, with_color
from slack.weechat_config import WeeChatOption, WeeChatOptionTypes
+if TYPE_CHECKING:
+ from typing_extensions import Literal
+
+ Options = Dict[str, Union[str, Literal[True]]]
+
REACTION_PREFIX_REGEX_STRING = (
rf"{MESSAGE_ID_REGEX_STRING}?{REACTION_CHANGE_REGEX_STRING}"
)
@@ -41,8 +46,8 @@ commands: Dict[str, Command] = {}
def parse_options(args: str):
regex = re.compile("(?:^| )+-([^ =]+)(?:=([^ ]+))?")
pos_args = regex.sub("", args)
- options: Dict[str, Optional[str]] = {
- match.group(1): match.group(2) for match in regex.finditer(args)
+ options: Options = {
+ match.group(1): match.group(2) or True for match in regex.finditer(args)
}
return pos_args, options
@@ -64,11 +69,11 @@ def weechat_command(
split_all_args: bool = False,
slack_buffer_required: bool = False,
) -> Callable[
- [Callable[[str, List[str], Dict[str, Optional[str]]], None]],
+ [Callable[[str, List[str], Options], None]],
Callable[[str, str], None],
]:
def decorator(
- f: Callable[[str, List[str], Dict[str, Optional[str]]], None],
+ f: Callable[[str, List[str], Options], None],
) -> Callable[[str, str], None]:
cmd = removeprefix(f.__name__, "command_").replace("_", " ")
top_level = " " not in cmd
@@ -116,7 +121,7 @@ def display_workspace(workspace: SlackWorkspace, detailed_list: bool):
@weechat_command()
-def command_slack(buffer: str, args: List[str], options: Dict[str, Optional[str]]):
+def command_slack(buffer: str, args: List[str], options: Options):
"""
slack command
"""
@@ -134,10 +139,8 @@ def workspace_connect(workspace: SlackWorkspace):
@weechat_command("%(slack_workspaces)|-all", split_all_args=True)
-def command_slack_connect(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
- if options.get("all", False) is None:
+def command_slack_connect(buffer: str, args: List[str], options: Options):
+ if options.get("all"):
for workspace in shared.workspaces.values():
run_async(workspace.connect())
elif args[0]:
@@ -161,10 +164,8 @@ def workspace_disconnect(workspace: SlackWorkspace):
@weechat_command("%(slack_workspaces)|-all", split_all_args=True)
-def command_slack_disconnect(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
- if options.get("all", False) is None:
+def command_slack_disconnect(buffer: str, args: List[str], options: Options):
+ if options.get("all"):
for workspace in shared.workspaces.values():
workspace.disconnect()
elif args[0]:
@@ -181,39 +182,29 @@ def command_slack_disconnect(
@weechat_command()
-def command_slack_rehistory(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_rehistory(buffer: str, args: List[str], options: Options):
slack_buffer = shared.buffers.get(buffer)
if slack_buffer:
run_async(slack_buffer.rerender_history())
@weechat_command()
-def command_slack_workspace(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_workspace(buffer: str, args: List[str], options: Options):
list_workspaces()
@weechat_command("%(slack_workspaces)")
-def command_slack_workspace_list(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_workspace_list(buffer: str, args: List[str], options: Options):
list_workspaces()
@weechat_command("%(slack_workspaces)")
-def command_slack_workspace_listfull(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_workspace_listfull(buffer: str, args: List[str], options: Options):
list_workspaces(detailed_list=True)
@weechat_command(min_args=1)
-def command_slack_workspace_add(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_workspace_add(buffer: str, args: List[str], options: Options):
name = args[0]
if name in shared.workspaces:
print_error(f'workspace "{name}" already exists, can\'t add it!')
@@ -226,7 +217,7 @@ def command_slack_workspace_add(
config_option: WeeChatOption[WeeChatOptionTypes] = getattr(
shared.workspaces[name].config, option_name
)
- value = "on" if option_value is None else option_value
+ value = "on" if option_value is True else option_value
config_option.value_set_as_str(value)
weechat.prnt(
@@ -236,9 +227,7 @@ def command_slack_workspace_add(
@weechat_command("%(slack_workspaces)", min_args=2)
-def command_slack_workspace_rename(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_workspace_rename(buffer: str, args: List[str], options: Options):
old_name = args[0]
new_name = args[1]
workspace = shared.workspaces.get(old_name)
@@ -256,9 +245,7 @@ def command_slack_workspace_rename(
@weechat_command("%(slack_workspaces)", min_args=1)
-def command_slack_workspace_del(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_workspace_del(buffer: str, args: List[str], options: Options):
name = args[0]
workspace = shared.workspaces.get(name)
if not workspace:
@@ -278,23 +265,18 @@ def command_slack_workspace_del(
@weechat_command("%(threads)", min_args=1)
-def command_slack_thread(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_thread(buffer: str, args: List[str], options: Options):
slack_buffer = shared.buffers.get(buffer)
if isinstance(slack_buffer, SlackConversation):
run_async(slack_buffer.open_thread(args[0], switch=True))
-def print_uncaught_error(
- error: UncaughtError, detailed: bool, options: Dict[str, Optional[str]]
-):
+def print_uncaught_error(error: UncaughtError, detailed: bool, options: Options):
weechat.prnt("", f" {error.id} ({error.time}): {error.exception}")
if detailed:
for line in format_exception(error.exception):
weechat.prnt("", f" {line}")
- data = options.get("data", False) is None
- if data:
+ if options.get("data"):
if isinstance(error.exception, SlackRtmError):
weechat.prnt("", f" data: {json.dumps(error.exception.message_json)}")
elif isinstance(error.exception, SlackError):
@@ -304,9 +286,7 @@ def print_uncaught_error(
@weechat_command("tasks|buffer|open_buffer|errors|error", split_all_args=True)
-def command_slack_debug(
- buffer: str, args: List[str], options: Dict[str, Optional[str]]
-):
+def command_slack_debug(buffer: str, args: List[str], options: Options):
# TODO: Add message info (message_json)
if args[0] == "tasks":
weechat.prnt("", "Active tasks:")