Skip to content

Commit

Permalink
Add py.typed and annotate most of the classes ant methods
Browse files Browse the repository at this point in the history
  • Loading branch information
[email protected] committed Feb 26, 2025
1 parent 9de8e88 commit 2e642da
Show file tree
Hide file tree
Showing 9 changed files with 319 additions and 167 deletions.
26 changes: 20 additions & 6 deletions i3ipc/_private/pubsub.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,23 @@
from typing import Callable, Optional, TypeAlias, TypedDict

from i3ipc.connection import Connection
from i3ipc.events import IpcBaseEvent

Handler: TypeAlias = Callable[[Connection, IpcBaseEvent], None]


class Subscription(TypedDict):
event: str
detail: Optional[str]
handler: Handler


class PubSub(object):
def __init__(self, conn):
def __init__(self, conn: Connection):
self.conn = conn
self._subscriptions = []
self._subscriptions: list[Subscription] = []

def subscribe(self, detailed_event, handler):
def subscribe(self, detailed_event: str, handler: Handler):
event = detailed_event.replace('-', '_')
detail = ''

Expand All @@ -12,10 +26,10 @@ def subscribe(self, detailed_event, handler):

self._subscriptions.append({'event': event, 'detail': detail, 'handler': handler})

def unsubscribe(self, handler):
def unsubscribe(self, handler: Handler):
self._subscriptions = list(filter(lambda s: s['handler'] != handler, self._subscriptions))

def emit(self, event, data):
def emit(self, event: str, data: Optional[IpcBaseEvent]):
detail = ''

if data and hasattr(data, 'change'):
Expand All @@ -27,4 +41,4 @@ def emit(self, event, data):
if data:
s['handler'](self.conn, data)
else:
s['handler'](self.conn)
s['handler'](self.conn) # type: ignore[call-arg]
6 changes: 3 additions & 3 deletions i3ipc/_private/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,19 +43,19 @@ class EventType(Enum):
TICK = (1 << 7)
INPUT = (1 << 21)

def to_string(self):
def to_string(self) -> str:
return str.lower(self.name)

@staticmethod
def from_string(val):
def from_string(val) -> 'EventType':
match = [e for e in EventType if e.to_string() == val]

if not match:
raise ValueError('event not implemented: ' + val)

return match[0]

def to_list(self):
def to_list(self) -> list[str]:
events_list = []
if self.value & EventType.WORKSPACE.value:
events_list.append(EventType.WORKSPACE.to_string())
Expand Down
Loading

0 comments on commit 2e642da

Please sign in to comment.