Adds py.typed and fixes mypy issues

i3ipc/aio/connection.py

@@ -6,7 +6,7 @@
 from .. import con
 import os
 import json
-from typing import Optional, List, Tuple, Callable, Union
+from typing import Optional, List, Set, Tuple, Callable, Union, cast
 import struct
 import socket
 import logging
@@ -137,7 +137,7 @@ class Con(con.Con):
     :ivar ipc_data: The raw data from the i3 ipc.
     :vartype ipc_data: dict
     """
-    async def command(self, command: str) -> List[CommandReply]:
+    async def command(self, command: str) -> List[CommandReply]: # type: ignore[override]
         """Runs a command on this container.
 
         .. seealso:: https://i3wm.org/docs/userguide.html#list_of_commands
@@ -146,9 +146,9 @@ async def command(self, command: str) -> List[CommandReply]:
             string.
         :rtype: list(CommandReply)
         """
-        return await self._conn.command('[con_id="{}"] {}'.format(self.id, command))
+        return await self._conn.command('[con_id="{}"] {}'.format(self.id, command)) # type: ignore[attr-defined]
 
-    async def command_children(self, command: str) -> List[CommandReply]:
+    async def command_children(self, command: str) -> List[CommandReply]: # type: ignore[override]
         """Runs a command on the immediate children of the currently selected
         container.
 
@@ -174,7 +174,7 @@ def _pack(msg_type: MessageType, payload: str) -> bytes:
 
 
 def _unpack_header(data: bytes) -> Tuple[bytes, int, int]:
-    return struct.unpack(_struct_header, data[:_struct_header_size])
+    return cast(Tuple[bytes, int, int], struct.unpack(_struct_header, data[:_struct_header_size]))
 
 
 async def _find_socket_path() -> Optional[str]:
@@ -258,9 +258,9 @@ def __init__(self, socket_path: Optional[str] = None, auto_reconnect: bool = Fal
         self._socket_path = socket_path
         self._auto_reconnect = auto_reconnect
         self._pubsub = _AIOPubSub(self)
-        self._subscriptions = set()
+        self._subscriptions: Set[Event] = set()
         self._main_future = None
-        self._reconnect_future = None
+        self._reconnect_future: Optional[Future] = None
         self._synchronizer = None
 
     def _sync(self):
@@ -270,7 +270,7 @@ def _sync(self):
         self._synchronizer.sync()
 
     @property
-    def socket_path(self) -> str:
+    def socket_path(self) -> Optional[str]:
         """The path of the socket this ``Connection`` is connected to.
 
         :rtype: str
@@ -487,7 +487,7 @@ async def subscribe(self, events: Union[List[Event], List[str]], force: bool = F
 
         for e in events:
             e = Event(e)
-            if e not in Event._subscribable_events:
+            if e not in cast(List[Event], Event._subscribable_events): # type: ignore[attr-defined]
                 correct_event = str.split(e.value, '::')[0].upper()
                 raise ValueError(
                     f'only nondetailed events are subscribable (use Event.{correct_event})')
@@ -512,7 +512,7 @@ async def subscribe(self, events: Union[List[Event], List[str]], force: bool = F
 
     def on(self,
            event: Union[Event, str],
-           handler: Callable[['Connection', IpcBaseEvent], None] = None):
+           handler: Optional[Callable[['Connection', IpcBaseEvent], None]] = None):
         def on_wrapped(handler):
             self._on(event, handler)
             return handler
@@ -531,7 +531,7 @@ def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBase
         :param handler: The event handler to call.
         :type handler: :class:`Callable`
         """
-        if type(event) is Event:
+        if isinstance(event, Event):
             event = event.value
 
         event = event.replace('-', '_')
@@ -594,7 +594,7 @@ async def get_bar_config_list(self) -> List[str]:
         data = await self._message(MessageType.GET_BAR_CONFIG)
         return json.loads(data)
 
-    async def get_bar_config(self, bar_id=None) -> Optional[BarConfigReply]:
+    async def get_bar_config(self, bar_id: Optional[str]=None) -> Optional[BarConfigReply]:
         """Gets the bar configuration specified by the id.
 
         :param bar_id: The bar id to get the configuration for. If not given,

i3ipc/con.py

@@ -181,7 +181,7 @@ def is_floating(self) -> bool:
         :returns: Whether this is a floating node
         :rtype: bool
         """
-        if self.floating in ['user_on', 'auto_on']:
+        if self.floating in ['user_on', 'auto_on']: # type: ignore[attr-defined]
             return True
         return False
 
@@ -249,7 +249,7 @@ def command(self, command: str) -> List[replies.CommandReply]:
             string.
         :rtype: list(:class:`CommandReply <i3ipc.CommandReply>`)
         """
-        return self._conn.command('[con_id="{}"] {}'.format(self.id, command))
+        return self._conn.command('[con_id="{}"] {}'.format(self.id, command)) # type: ignore[attr-defined]
 
     def command_children(self, command: str) -> List[replies.CommandReply]:
         """Runs a command on the immediate children of the currently selected
@@ -261,13 +261,13 @@ def command_children(self, command: str) -> List[replies.CommandReply]:
         :rtype: list(:class:`CommandReply <i3ipc.CommandReply>`)
         """
         if not len(self.nodes):
-            return
+            return []
 
         commands = []
         for c in self.nodes:
             commands.append('[con_id="{}"] {};'.format(c.id, command))
 
-        self._conn.command(' '.join(commands))
+        return self._conn.command(' '.join(commands))
 
     def workspaces(self) -> List['Con']:
         """Gets a list of workspace containers for this tree.
@@ -393,8 +393,8 @@ def find_marked(self, pattern: str = ".*") -> List['Con']:
             pattern.
         :rtype: list(:class:`Con`)
         """
-        pattern = re.compile(pattern)
-        return [c for c in self if any(pattern.search(mark) for mark in c.marks)]
+        compiled_pattern = re.compile(pattern)
+        return [c for c in self if any(compiled_pattern.search(mark) for mark in c.marks)]
 
     def find_fullscreen(self) -> List['Con']:
         """Finds all the containers under this node that are in fullscreen
@@ -425,7 +425,7 @@ def workspace(self) -> Optional['Con']:
 
         return ret
 
-    def scratchpad(self) -> 'Con':
+    def scratchpad(self) -> Optional['Con']:
         """Finds the scratchpad container.
 
         :returns: The scratchpad container.

i3ipc/connection.py

@@ -232,7 +232,7 @@ def get_version(self) -> VersionReply:
         data = json.loads(data)
         return VersionReply(data)
 
-    def get_bar_config(self, bar_id: str = None) -> Optional[BarConfigReply]:
+    def get_bar_config(self, bar_id: Optional[str] = None) -> Optional[BarConfigReply]:
         """Gets the bar configuration specified by the id.
 
         :param bar_id: The bar id to get the configuration for. If not given,
@@ -391,7 +391,7 @@ def off(self, handler: Callable[['Connection', IpcBaseEvent], None]):
 
     def on(self,
            event: Union[Event, str],
-           handler: Callable[['Connection', IpcBaseEvent], None] = None):
+           handler: Optional[Callable[['Connection', IpcBaseEvent], None]] = None):
         def on_wrapped(handler):
             self._on(event, handler)
             return handler
@@ -410,7 +410,7 @@ def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBase
         :param handler: The event handler to call.
         :type handler: :class:`Callable`
         """
-        if type(event) is Event:
+        if isinstance(event, Event):
             event = event.value
 
         event = event.replace('-', '_')
@@ -426,7 +426,7 @@ def _on(self, event: Union[Event, str], handler: Callable[['Connection', IpcBase
             self._pubsub.subscribe(event, handler)
             return
 
-        event_type = 0
+        event_type: Optional[EventType] = None
         if base_event == 'workspace':
             event_type = EventType.WORKSPACE
         elif base_event == 'output':

i3ipc/events.py

@@ -45,7 +45,7 @@ class Event(Enum):
     INPUT_REMOVED = 'input::removed'
 
 
-Event._subscribable_events = [e for e in Event if '::' not in e.value]
+Event._subscribable_events = [e for e in Event if '::' not in e.value] # type: ignore[attr-defined]
 
 
 class WorkspaceEvent(IpcBaseEvent):

setup.py

@@ -34,6 +34,7 @@
     install_requires=REQUIRED,
     extras_require=EXTRAS,
     include_package_data=True,
+    package_data={"i3ipc": ["py.typed"]},
     license='BSD',
     keywords='i3 i3wm extensions add-ons',
     classifiers=[

test/aio/test_shutdown_event.py

@@ -1,3 +1,7 @@
+from typing import List
+from i3ipc.aio.connection import Connection
+
+from i3ipc.events import IpcBaseEvent, ShutdownEvent
 from .ipctest import IpcTest
 
 import pytest
@@ -6,20 +10,21 @@
 
 
 class TestShutdownEvent(IpcTest):
-    events = []
+    events: List[ShutdownEvent] = []
 
-    def restart_func(self, i3):
+    def restart_func(self, i3: Connection):
         asyncio.ensure_future(i3.command('restart'))
 
-    def on_shutdown(self, i3, e):
+    def on_shutdown(self, i3: Connection, e: IpcBaseEvent):
+        assert isinstance(e, ShutdownEvent)
         self.events.append(e)
         if len(self.events) == 1:
             i3._loop.call_later(0.1, self.restart_func, i3)
         elif len(self.events) == 2:
             i3.main_quit()
 
     @pytest.mark.asyncio
-    async def test_shutdown_event_reconnect(self, i3):
+    async def test_shutdown_event_reconnect(self, i3: Connection):
         i3._auto_reconnect = True
         self.events = []
         i3.on('shutdown::restart', self.on_shutdown)

test/aio/test_ticks.py

@@ -1,11 +1,14 @@
+from typing import List
+
+from i3ipc.events import TickEvent
 from .ipctest import IpcTest
 
 import pytest
 import asyncio
 
 
 class TestTicks(IpcTest):
-    events = []
+    events: List[TickEvent] = []
 
     async def on_tick(self, i3, e):
         self.events.append(e)

test/test_shutdown_event.py

@@ -1,14 +1,16 @@
 from threading import Timer
+from typing import List
+from i3ipc.events import ShutdownEvent
 from ipctest import IpcTest
 
 
 class TestShutdownEvent(IpcTest):
-    events = []
+    events: List[ShutdownEvent] = []
 
     def restart_func(self, i3):
         i3.command('restart')
 
-    def on_shutdown(self, i3, e):
+    def on_shutdown(self, i3, e: ShutdownEvent):
         self.events.append(e)
         assert i3._wait_for_socket()
         if len(self.events) == 1:

test/test_ticks.py

@@ -1,10 +1,12 @@
+from typing import List
+from i3ipc.events import TickEvent
 from ipctest import IpcTest
 
 
 class TestTicks(IpcTest):
-    events = []
+    events: List[TickEvent] = []
 
-    def on_tick(self, i3, e):
+    def on_tick(self, i3, e: TickEvent):
         self.events.append(e)
         if len(self.events) == 3:
             i3.main_quit()