aboutsummaryrefslogtreecommitdiffstats
path: root/slack
diff options
context:
space:
mode:
Diffstat (limited to 'slack')
-rw-r--r--slack/commands.py6
-rw-r--r--slack/shared.py10
-rw-r--r--slack/slack_workspace.py12
-rw-r--r--slack/task.py6
-rw-r--r--slack/util.py8
-rw-r--r--slack/weechat_config.py3
6 files changed, 23 insertions, 22 deletions
diff --git a/slack/commands.py b/slack/commands.py
index c60b420..8dfce84 100644
--- a/slack/commands.py
+++ b/slack/commands.py
@@ -4,7 +4,7 @@ import pprint
import re
from dataclasses import dataclass
from functools import wraps
-from typing import Any, Callable, Dict, List, Optional, Tuple
+from typing import Callable, Dict, List, Optional, Tuple
import weechat
@@ -19,7 +19,7 @@ from slack.slack_user import name_from_user_info_without_spaces
from slack.slack_workspace import SlackWorkspace
from slack.task import create_task
from slack.util import get_callback_name, with_color
-from slack.weechat_config import WeeChatOption
+from slack.weechat_config import WeeChatOption, WeeChatOptionTypes
commands: Dict[str, Command] = {}
@@ -164,7 +164,7 @@ def command_slack_workspace_add(
for option_name, option_value in options.items():
if hasattr(shared.workspaces[name].config, option_name):
- config_option: WeeChatOption[Any] = getattr(
+ config_option: WeeChatOption[WeeChatOptionTypes] = getattr(
shared.workspaces[name].config, option_name
)
value = "on" if option_value is None else option_value
diff --git a/slack/shared.py b/slack/shared.py
index 1e9b2aa..31ff748 100644
--- a/slack/shared.py
+++ b/slack/shared.py
@@ -1,13 +1,15 @@
from __future__ import annotations
from collections import defaultdict
-from typing import TYPE_CHECKING, Any, Dict, List
+from typing import TYPE_CHECKING, Callable, Dict, List, Union
if TYPE_CHECKING:
from slack.config import SlackConfig
from slack.slack_workspace import SlackWorkspace
from slack.task import Future, Task
+WeechatCallbackReturnType = Union[int, str, Dict[str, str], None]
+
class Shared:
def __init__(self):
@@ -15,9 +17,9 @@ class Shared:
self.SCRIPT_VERSION = "3.0.0"
self.weechat_version: int
- self.weechat_callbacks: Dict[str, Any]
- self.active_tasks: Dict[str, List[Task[Any]]] = defaultdict(list)
- self.active_futures: Dict[str, Future[Any]] = {}
+ self.weechat_callbacks: Dict[str, Callable[..., WeechatCallbackReturnType]]
+ self.active_tasks: Dict[str, List[Task[object]]] = defaultdict(list)
+ self.active_futures: Dict[str, Future[object]] = {}
self.workspaces: Dict[str, SlackWorkspace] = {}
self.config: SlackConfig
diff --git a/slack/slack_workspace.py b/slack/slack_workspace.py
index 176e031..ac2b1e7 100644
--- a/slack/slack_workspace.py
+++ b/slack/slack_workspace.py
@@ -5,7 +5,7 @@ import socket
import ssl
import time
from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Any, Dict, Generic, Iterable, Optional, Type, TypeVar
+from typing import TYPE_CHECKING, Dict, Generic, Iterable, Optional, Type, TypeVar
import weechat
from websocket import ABNF, WebSocketConnectionClosedException, create_connection
@@ -24,10 +24,10 @@ if TYPE_CHECKING:
from slack_api.slack_usergroups_info import SlackUsergroupInfo
from slack_api.slack_users_info import SlackUserInfo
else:
- SlackBotInfo = Any
- SlackConversationsInfo = Any
- SlackUsergroupInfo = Any
- SlackUserInfo = Any
+ SlackBotInfo = object
+ SlackConversationsInfo = object
+ SlackUsergroupInfo = object
+ SlackUserInfo = object
SlackItemClass = TypeVar(
"SlackItemClass", SlackConversation, SlackUser, SlackBot, SlackUsergroup
@@ -247,5 +247,5 @@ class SlackWorkspace:
self.ws_recv(json.loads(recv_data.decode()))
- def ws_recv(self, data: Any):
+ def ws_recv(self, data: object):
print(f"received: {data}")
diff --git a/slack/task.py b/slack/task.py
index 763f605..0209be4 100644
--- a/slack/task.py
+++ b/slack/task.py
@@ -80,7 +80,7 @@ class Task(Future[T]):
return f"{self.__class__.__name__}('{self.id}', coroutine={self.coroutine.__qualname__})"
-def weechat_task_cb(data: str, *args: Any) -> int:
+def weechat_task_cb(data: str, *args: object) -> int:
future = shared.active_futures.pop(data)
future.set_result(args)
tasks = shared.active_tasks.pop(data)
@@ -89,7 +89,7 @@ def weechat_task_cb(data: str, *args: Any) -> int:
return weechat.WEECHAT_RC_OK
-def process_ended_task(task: Task[Any], response: Any):
+def process_ended_task(task: Task[Any], response: object):
task.set_result(response)
if task.id in shared.active_tasks:
tasks = shared.active_tasks.pop(task.id)
@@ -99,7 +99,7 @@ def process_ended_task(task: Task[Any], response: Any):
del shared.active_futures[task.id]
-def task_runner(task: Task[Any], response: Any):
+def task_runner(task: Task[Any], response: object):
while True:
try:
future = task.coroutine.send(response)
diff --git a/slack/util.py b/slack/util.py
index 8738748..8dfc967 100644
--- a/slack/util.py
+++ b/slack/util.py
@@ -1,15 +1,13 @@
from __future__ import annotations
-from typing import Callable, Dict, Union
+from typing import Callable
import weechat
-from slack.shared import shared
+from slack.shared import WeechatCallbackReturnType, shared
-weechat_callback_return_type = Union[int, str, Dict[str, str], None]
-
-def get_callback_name(callback: Callable[..., weechat_callback_return_type]) -> str:
+def get_callback_name(callback: Callable[..., WeechatCallbackReturnType]) -> str:
callback_id = f"{callback.__name__}-{id(callback)}"
shared.weechat_callbacks[callback_id] = callback
return callback_id
diff --git a/slack/weechat_config.py b/slack/weechat_config.py
index 6decf37..8d17b13 100644
--- a/slack/weechat_config.py
+++ b/slack/weechat_config.py
@@ -48,7 +48,8 @@ class WeeChatSection:
)
-WeeChatOptionType = TypeVar("WeeChatOptionType", bound=Union[int, str])
+WeeChatOptionTypes = Union[int, str]
+WeeChatOptionType = TypeVar("WeeChatOptionType", bound=WeeChatOptionTypes)
@dataclass