GitLab

at 2021-12-06T23:51:45+01:00

Observable: Remove unused code

Observable: Add type annotations

App: Add type annotations

Remove unused code

BaseModule: Add type annotations

 from __future__ import annotations

-from typing import Any  # pylint: disable=unused-import

-from typing import Dict  # pylint: disable=unused-import

-from typing import List  # pylint: disable=unused-import

-from typing import Optional  # pylint: disable=unused-import

+from typing import Any

+from typing import NamedTuple

+from typing import Union

+from typing import Optional

 from typing import cast

 import gc

 import logging

 import weakref

 import pprint

-from collections import namedtuple

 from collections import defaultdict

 import nbxmpp

 from nbxmpp.idlequeue import IdleQueue

 from gi.repository import Gdk

 from gi.repository import GLib

+from nbxmpp.protocol import JID

 import gajim

+from gajim.common import types

 from gajim.common import config as c_config

 from gajim.common import configpaths

 from gajim.common import ged as ged_module

 from gajim.common.i18n import LANG

 from gajim.common.const import Display

-from gajim.common.types import NetworkEventsControllerT  # pylint: disable=unused-import

-from gajim.common.types import InterfaceT  # pylint: disable=unused-import

-from gajim.common.types import ConnectionT  # pylint: disable=unused-import

-from gajim.common.types import LegacyContactsAPIT  # pylint: disable=unused-import

-from gajim.common.types import SettingsT  # pylint: disable=unused-import

-from gajim.common.types import CSSConfigT  # pylint: disable=unused-import

-interface = cast(InterfaceT, None)

+interface = cast(types.InterfaceT, None)

 thread_interface = lambda *args: None # Interface to run a thread and then a callback

 config = c_config.Config()

-settings = cast(SettingsT, None)

+settings = cast(types.SettingsT, None)

 version = gajim.__version__

-connections = {}  # type: Dict[str, ConnectionT]

-avatar_cache = {}  # type: Dict[str, Dict[str, Any]]

-bob_cache = {} # type: Dict[str, bytes]

+connections: dict[str, types.Client] = {}

+avatar_cache: dict[str, dict[str, Any]] = {}

+bob_cache: dict[str, bytes] = {}

 ipython_window = None

 app = None  # Gtk.Application

 window = None # MainWindow

 ged = ged_module.GlobalEventsDispatcher() # Global Events Dispatcher

-nec = cast(NetworkEventsControllerT, None)

+nec = cast(types.NetworkEventsControllerT, None)

 plugin_manager = None # Plugins Manager

 class Storage:

 storage = Storage()

-css_config = cast(CSSConfigT, None)

+css_config = cast(types.CSSConfigT, None)

-transport_type = {}  # type: Dict[str, str]

-

-contacts = cast(LegacyContactsAPIT, None)

-

-# tell if we are connected to the room or not

-# {acct: {room_jid: True}}

-gc_connected = {}  # type: Dict[str, Dict[str, bool]]

+transport_type: dict[str, str] = {}

 # dict of rooms that must be automatically configured

 # and for which we have a list of invities

 # {account: {room_jid: {'invities': []}}}

-automatic_rooms = {}  # type: Dict[str, Dict[str, Dict[str, List[str]]]]

-

- # dict of groups, holds if they are expanded or not

-groups = {}  # type: Dict[str, Dict[str, Dict[str, bool]]]

+automatic_rooms: dict[str, dict[str, dict[str, list[str]]]] = {}

 # list of contacts that has just signed in

-newly_added = {}  # type: Dict[str, List[str]]

+newly_added: dict[str, list[str]] = {}

 # list of contacts that has just signed out

-to_be_removed = {}  # type: Dict[str, List[str]]

+to_be_removed: dict[str, list[str]] = {}

 notification = None

 # list of our nick names in each account

-nicks = {}  # type: Dict[str, str]

+nicks: dict[str, str] = {}

 # should we block 'contact signed in' notifications for this account?

 # this is only for the first 30 seconds after we change our show

 # to something else than offline

 # can also contain account/transport_jid to block notifications for contacts

 # from this transport

-block_signed_in_notifications = {}  # type: Dict[str, bool]

+block_signed_in_notifications: dict[str, bool] = {}

 proxy65_manager = None

     'AYATANA_APPINDICATOR': False,

 }

-_tasks = defaultdict(list)  # type: Dict[int, List[Any]]

+_tasks: dict[int, list[Any]] = defaultdict(list)

+

+

+class RecentGroupchat(NamedTuple):

+    room: str

+    server: str

+    nickname: str

-def print_version():

+

+def print_version() -> None:

     log('gajim').info('Gajim Version: %s', gajim.__version__)

-def get_client(account):

+def get_client(account: str) -> types.Client:

     return connections[account]

-def is_installed(dependency):

+def is_installed(dependency: str) -> bool:

     if dependency == 'ZEROCONF':

         # Alias for checking zeroconf libs

         return _dependencies['AVAHI'] or _dependencies['PYBONJOUR']

     return _dependencies[dependency]

-def is_flatpak():

+def is_flatpak() -> bool:

     return gajim.IS_FLATPAK

-def is_portable():

+def is_portable() -> bool:

     return gajim.IS_PORTABLE

-def is_display(display):

+def is_display(display: Display) -> bool:

     # XWayland reports as Display X11, so try with env var

     is_wayland = os.environ.get('XDG_SESSION_TYPE') == 'wayland'

     if is_wayland and display == Display.WAYLAND:

         return False

     return default.__class__.__name__ == display.value

-def disable_dependency(dependency):

+

+def disable_dependency(dependency: str) -> None:

     _dependencies[dependency] = False

-def detect_dependencies():

+

+def detect_dependencies() -> None:

     import gi

     # ZEROCONF

     log('gajim').info('Used language: %s', LANG)

-def detect_desktop_env():

+

+def detect_desktop_env() -> Optional[str]:

     if sys.platform in ('win32', 'darwin'):

         return sys.platform

 desktop_env = detect_desktop_env()

-def get_server_from_jid(jid):

+def get_server_from_jid(jid: str) -> str:

     pos = jid.find('@') + 1 # after @

     return jid[pos:]

-def get_room_and_nick_from_fjid(jid):

+def get_room_and_nick_from_fjid(jid: str) -> list[str]:

     # fake jid is the jid for a contact in a room

     # gaim@conference.jabber.no/nick/nick-continued

     # return ('gaim@conference.jabber.no', 'nick/nick-continued')

         l.append('')

     return l

-def get_jid_without_resource(jid):

+

+def get_jid_without_resource(jid: str) -> str:

     return jid.split('/')[0]

-def get_number_of_accounts():

+def get_number_of_accounts() -> int:

     """

     Return the number of ALL accounts

     """

     return len(connections.keys())

-def get_number_of_connected_accounts(accounts_list=None):

+

+def get_number_of_connected_accounts(

+        accounts_list: Optional[list[str]] = None) -> int:

     """

-    Returns the number of CONNECTED accounts. Uou can optionally pass an

+    Returns the number of connected accounts. You can optionally pass an

     accounts_list and if you do those will be checked, else all will be checked

     """

     connected_accounts = 0

             connected_accounts = connected_accounts + 1

     return connected_accounts

-def get_available_clients():

-    clients = []

+

+def get_available_clients() -> list[types.Client]:

+    clients: list[types.Client] = []

     for client in connections.values():

         if client.state.is_available:

             clients.append(client)

     return clients

-def get_connected_accounts(exclude_local=False):

+

+def get_connected_accounts(exclude_local: bool = False) -> list[str]:

     """

     Returns a list of CONNECTED accounts

     """

-    account_list = []

+    account_list: list[str] = []

     for account in connections:

         if account == 'Local' and exclude_local:

             continue

             account_list.append(account)

     return account_list

-def get_accounts_sorted():

+

+def get_accounts_sorted() -> list[str]:

     '''

     Get all accounts alphabetically sorted with Local first

     '''

         account_list.insert(0, 'Local')

     return account_list

-def get_enabled_accounts_with_labels(exclude_local=True, connected_only=False,

-                                     private_storage_only=False):

+

+def get_enabled_accounts_with_labels(

+        exclude_local: bool = True,

+        connected_only: bool = False,

+        private_storage_only: bool = False) -> list[str]:

     """

     Returns a list with [account, account_label] entries.

     Order by account_label

     """

-    accounts = []

+    accounts: list[str] = []

     for acc in connections:

         if exclude_local and account_is_zeroconf(acc):

             continue

     accounts.sort(key=lambda xs: str.lower(xs[1]))

     return accounts

-def get_account_label(account):

+

+def get_account_label(account: str) -> str:

     return settings.get_account_setting(account, 'account_label') or account

-def account_is_zeroconf(account):

+

+def account_is_zeroconf(account: str) -> bool:

     return connections[account].is_zeroconf

-def account_supports_private_storage(account):

+

+def account_supports_private_storage(account: str) -> bool:

     # If Delimiter module is not available we can assume

     # Private Storage is not available

     return connections[account].get_module('Delimiter').available

-def account_is_connected(account):

+

+def account_is_connected(account: str) -> bool:

     if account not in connections:

         return False

     return (connections[account].state.is_connected or

             connections[account].state.is_available)

-def account_is_available(account):

+

+def account_is_available(account: str) -> bool:

     if account not in connections:

         return False

     return connections[account].state.is_available

-def account_is_disconnected(account):

+

+def account_is_disconnected(account: str) -> bool:

     return not account_is_connected(account)

-def zeroconf_is_connected():

+

+def zeroconf_is_connected() -> bool:

     return account_is_connected(ZEROCONF_ACC_NAME) and \

             settings.get_account_setting(ZEROCONF_ACC_NAME, 'is_zeroconf')

-def in_groupchat(account, room_jid):

-    room_jid = str(room_jid)

-    if room_jid not in gc_connected[account]:

-        return False

-    return gc_connected[account][room_jid]

-def get_transport_name_from_jid(jid, use_config_setting=True):

+def get_transport_name_from_jid(

+        jid: str,

+        use_config_setting: bool = True) -> Optional[str]:

+

     """

     Returns 'gg', 'irc' etc

         return 'facebook'

     return None

-def jid_is_transport(jid):

+def jid_is_transport(jid: str) -> bool:

     # if not '@' or '@' starts the jid then it is transport

     if jid.find('@') <= 0:

         return True

     return False

-def get_jid_from_account(account_name):

+

+def get_jid_from_account(account_name: str) -> str:

     """

     Return the jid we use in the given account

     """

     jid = name + '@' + hostname

     return jid

-def get_account_from_jid(jid):

+

+def get_account_from_jid(jid: str) -> Optional[str]:

     for account in settings.get_accounts():

         if jid == get_jid_from_account(account):

             return account

-def get_hostname_from_account(account_name, use_srv=False):

+

+def get_hostname_from_account(account_name: str, use_srv: bool = False) -> str:

     """

     Returns hostname (if custom hostname is used, that is returned)

     """

         return settings.get_account_setting(account_name, 'custom_host')

     return settings.get_account_setting(account_name, 'hostname')

-def get_notification_image_prefix(jid):

+

+def get_notification_image_prefix(jid: str) -> str:

     """

     Returns the prefix for the notification images

     """

     transport_name = get_transport_name_from_jid(jid)

-    if transport_name in ('icq', 'facebook'):

+    if transport_name in ['icq', 'facebook']:

         prefix = transport_name

     else:

         prefix = 'jabber'

     return prefix

-def get_recent_groupchats(account):

+def get_recent_groupchats(account: str) -> list[RecentGroupchat]:

     recent_groupchats = settings.get_account_setting(

         account, 'recent_groupchats').split()

-    RecentGroupchat = namedtuple('RecentGroupchat',

-                                 ['room', 'server', 'nickname'])

-

-    recent_list = []

+    recent_list: list[RecentGroupchat] = []

     for groupchat in recent_groupchats:

         jid = nbxmpp.JID.from_string(groupchat)

         recent = RecentGroupchat(jid.localpart, jid.domain, jid.resource)

         recent_list.append(recent)

     return recent_list

-def add_recent_groupchat(account, room_jid, nickname):

+

+def add_recent_groupchat(account: str, room_jid: str, nickname: str) -> None:

     recent = settings.get_account_setting(

         account, 'recent_groupchats').split()

     full_jid = room_jid + '/' + nickname

     config_value = ' '.join(recent)

     settings.set_account_setting(account, 'recent_groupchats', config_value)

-def get_priority(account, show):

+

+def get_priority(account: str, show: str) -> int:

     """

     Return the priority an account must have

     """

         prio = 127

     return prio

-def log(domain):

+

+def log(domain: str) -> logging.Logger:

     if domain != 'gajim':

         domain = 'gajim.%s' % domain

     return logging.getLogger(domain)

-def prefers_app_menu():

+

+def prefers_app_menu() -> bool:

     if sys.platform == 'darwin':

         return True

     if sys.platform == 'win32':

         return False

     return app.prefers_app_menu()

-def load_css_config():

+

+def load_css_config() -> None:

     global css_config

     from gajim.gui.css_config import CSSConfig

     css_config = CSSConfig()

+

 def set_debug_mode(enable: bool) -> None:

     debug_folder = configpaths.get('DEBUG')

     debug_enabled = debug_folder / 'debug-enabled'

         if debug_enabled.exists():

             debug_enabled.unlink()

+

 def get_debug_mode() -> bool:

     debug_folder = configpaths.get('DEBUG')

     debug_enabled = debug_folder / 'debug-enabled'

     return debug_enabled.exists()

+

 def get_stored_bob_data(algo_hash: str) -> Optional[bytes]:

     try:

         return bob_cache[algo_hash]

     _tasks[id(self)].append(task)

-def remove_task(task, id_):

+def remove_task(task, id_) -> None:

     try:

         _tasks[id_].remove(task)

     except Exception:

             del _tasks[id_]

-def cancel_tasks(obj):

+def cancel_tasks(obj: Any) -> None:

     id_ = id(obj)

     if id_ not in _tasks:

         return

         task.cancel()

-def check_finalize(obj):

+def check_finalize(obj: Any) -> None:

     if 'GAJIM_LEAK' not in os.environ:

         return

+

     name = obj.__class__.__name__

     logger = logging.getLogger('gajim.leak')

     finalizer = weakref.finalize(obj, logger.info, f'{name} has been finalized')

 # You should have received a copy of the GNU General Public License

 # along with Gajim. If not, see <http://www.gnu.org/licenses/>.

-from typing import Any  # pylint: disable=unused-import

-from typing import Dict  # pylint: disable=unused-import

+from __future__ import annotations

+

+from typing import Any

+from typing import Dict

 from typing import Optional

 from typing import List

 from typing import Tuple

 from gajim.common.regex import INVALID_XML_CHARS_REGEX

 from gajim.common.regex import STH_AT_STH_DOT_STH_REGEX

 from gajim.common.structs import URI

+from gajim.common import types

 HAS_PYWIN32 = False

 if os.name == 'nt':

 class Observable:

-    def __init__(self, log_=None):

+    def __init__(self, log_: Optional[logging.Logger] = None):

         self._log = log_

-        self._callbacks = defaultdict(lambda: defaultdict(list))

+        self._callbacks: types.ObservableCbDict = defaultdict(list)

-    def disconnect_signals(self):

-        self._callbacks = defaultdict(lambda: defaultdict(list))

+    def disconnect_signals(self) -> None:

+        self._callbacks = defaultdict(list)

-    def disconnect_all_from_obj(self, object_):

-        for qualifiers in self._callbacks.values():

-            for handlers in qualifiers.values():

-                for handler in list(handlers):

-                    func = handler()

-                    if func is None or func.__self__ is object_:

-                        handlers.remove(handler)

+    def disconnect_all_from_obj(self, obj: Any) -> None:

+        for handlers in self._callbacks.values():

+            for handler in list(handlers):

+                func = handler()

+                if func is None or func.__self__ is obj:

+                    handlers.remove(handler)

-    def disconnect(self, *args):

-        self.disconnect_all_from_obj(*args)

+    def disconnect(self, obj: Any) -> None:

+        self.disconnect_all_from_obj(obj)

-    def connect_signal(self, signal_name, func, qualifiers=None):

+    def connect_signal(self,

+                       signal_name: str,

+                       func: types.AnyCallableT) -> None:

         if not inspect.ismethod(func):

             raise ValueError('Only bound methods allowed')

         weak_func = weakref.WeakMethod(func)

-        self._callbacks[signal_name][qualifiers].append(weak_func)

+        self._callbacks[signal_name].append(weak_func)

-    def connect(self, *args, **kwargs):

-        self.connect_signal(*args, **kwargs)

+    def connect(self,

+                signal_name: str,

+                func: types.AnyCallableT) -> None:

+        self.connect_signal(signal_name, func)

-    def multi_connect(self, signal_dict):

+    def multi_connect(self, signal_dict: dict[str, types.AnyCallableT]):

         for signal_name, func in signal_dict.items():

             self.connect_signal(signal_name, func)

-    def notify(self, signal_name, *args, qualifiers=None, **kwargs):

+    def notify(self, signal_name: str, *args: Any, **kwargs: Any):

         signal_callbacks = self._callbacks.get(signal_name)

-        if signal_callbacks is None:

+        if not signal_callbacks:

             return

         if self._log is not None:

             self._log.info('Signal: %s', signal_name)

-        callbacks = signal_callbacks.get(qualifiers, [])

-        for func in list(callbacks):

-            if func() is None:

-                self._callbacks[signal_name][qualifiers].remove(func)

+        for weak_method in list(signal_callbacks):

+            func = weak_method()

+            if func is None:

+                self._callbacks[signal_name].remove(weak_method)

                 continue

-            func()(self, signal_name, *args, **kwargs)

+            func(self, signal_name, *args, **kwargs)

 def write_file_async(path, data, callback, user_data=None):

 # You should have received a copy of the GNU General Public License

 # along with Gajim.  If not, see <http://www.gnu.org/licenses/>.

+from __future__ import annotations

+

 from typing import Any

-from typing import Dict

-from typing import List

+from typing import Optional

 import logging

 from functools import partial

 from nbxmpp.namespaces import Namespace

 from nbxmpp.structs import StanzaHandler

+from gajim.common import types

 from gajim.common import app

 from gajim.common.nec import EventHelper

 from gajim.common.modules.util import LogAdapter

 class BaseModule(EventHelper):

     _nbxmpp_extends = ''

-    _nbxmpp_methods = []  # type: List[str]

+    _nbxmpp_methods: list[str] = []

+

+    def __init__(self,

+                 con: types.Client,

+                 *args: Any,

+                 plugin: bool = False,

+                 **kwargs: Any):

-    def __init__(self, con, *args, plugin=False, **kwargs):

         EventHelper.__init__(self)

         self._con = con

-        self._account = con.name

+        self._account = con.account

         self._log = self._set_logger(plugin)

-        self._nbxmpp_callbacks: Dict[str, Any] = {}

-        self._stored_publish = None  # type: Callable

-        self.handlers: List[StanzaHandler] = []

+        self._nbxmpp_callbacks: dict[str, Any] = {}

+        self._stored_publish: Optional[types.AnyCallableT] = None

+        self.handlers: list[StanzaHandler] = []

-    def _set_logger(self, plugin):

+    def _set_logger(self, plugin: bool) -> LogAdapter:

         logger_name = 'gajim.c.m.%s'

         if plugin:

             logger_name = 'gajim.p.%s'

         logger = logging.getLogger(logger_name)

         return LogAdapter(logger, {'account': self._account})

-    def __getattr__(self, key):

+    def __getattr__(self, key: str) -> Any:

         if key not in self._nbxmpp_methods:

             raise AttributeError(

                 "attribute '%s' is neither part of object '%s' "

             return getattr(module, key)

         return partial(getattr(module, key), callback=callback)

-    def _nbxmpp(self, module_name=None):

+    def _nbxmpp(self, module_name: Optional[str] = None):

         if not app.account_is_connected(self._account):

             self._log.warning('Account not connected, can’t use nbxmpp method')

             return Mock()

     def _register_callback(self, method, callback):

         self._nbxmpp_callbacks[method] = callback

-    def _register_pubsub_handler(self, callback):

+    def _register_pubsub_handler(self, callback: types.AnyCallableT):

         handler = StanzaHandler(name='message',

                                 callback=callback,

                                 ns=Namespace.PUBSUB_EVENT,

                                 priority=49)

         self.handlers.append(handler)

-    def send_stored_publish(self):

+    def send_stored_publish(self) -> None:

         if self._stored_publish is None:

             return

         self._log.info('Send stored publish')

         self._stored_publish()  # pylint: disable=not-callable

-    def _get_contact(self, jid, groupchat=False):

+    def _get_contact(self, jid, groupchat: bool = False):

         return self._con.get_module('Contacts').get_contact(

             jid, groupchat=groupchat)

-    def cleanup(self):

+    def cleanup(self) -> None:

         self.unregister_events()

             self._log.info('Invite from: %s, to: %s', data.from_, data.muc)

-            if app.in_groupchat(self._account, data.muc):

+            contact = self._get_contact(data.muc, groupchat=True)

+            if contact.is_joined:

                 # We are already in groupchat. Ignore invitation

                 self._log.info('We are already in this room')

                 raise nbxmpp.NodeProcessed

 # Types for typechecking

+from typing import Any

 from typing import Callable

 from typing import Dict

 from typing import List

 from typing import TYPE_CHECKING

 from pathlib import Path

+import weakref

 from gi.repository import GdkPixbuf

 ConnectionT = Union['Client', 'ConnectionZeroconf']

 ContactsT = Union['Contact', 'GC_Contact']

 ContactT = Union['Contact']

-LegacyContactsAPIT = Union['LegacyContactsAPI']

 CSSConfigT = Union['CSSConfig']

 # PEP

 BookmarksDict = Dict[JID, BookmarkData]

 GdkPixbufType = Union[GdkPixbuf.Pixbuf, GdkPixbuf.PixbufAnimation]

+

+AnyCallableT = Callable[..., Any]

+ObservableCbDict = dict[str, list[weakref.WeakMethod[AnyCallableT]]]
                 'infos': {},

                 'disco': {}

             }

-            app.groups[acc] = {}

-            app.gc_connected[acc] = {}

             app.automatic_rooms[acc] = {}

             app.newly_added[acc] = []

             app.to_be_removed[acc] = []

             'disco': {}

         }

-        app.groups[account] = {}

-        app.gc_connected[account] = {}

         app.automatic_rooms[account] = {}

         app.newly_added[account] = []

         app.to_be_removed[account] = []

         del self.instances[account]

         del app.nicks[account]

         del app.block_signed_in_notifications[account]

-        del app.groups[account]

-        del app.gc_connected[account]

         del app.automatic_rooms[account]

         del app.to_be_removed[account]

         del app.newly_added[account]

Philipp Hörist pushed to branch master at gajim / gajim

Commits:

6 changed files:

Changes: