diff options
| -rw-r--r-- | slack/commands.py | 6 | ||||
| -rw-r--r-- | slack/shared.py | 10 | ||||
| -rw-r--r-- | slack/slack_workspace.py | 12 | ||||
| -rw-r--r-- | slack/task.py | 6 | ||||
| -rw-r--r-- | slack/util.py | 8 | ||||
| -rw-r--r-- | slack/weechat_config.py | 3 |
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 |