feat: type annotations (#165)

* feat: type annotations

* fix tests accordingly

* cast to timezone

* guard more imports on icalendar version

* remove bad import from merge resolve

* chore: apply suggestions

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

---------

Co-authored-by: Lev Blit <levblt1@gmail.com>
Co-authored-by: Martin Eigenmann <eigenmannmartin@users.noreply.github.com>

Author: 3 people

MANIFEST.in

@@ -1 +1 @@
-recursive-include test *.ics *.txt
+recursive-include test *.ics *.txt py.typed

icalevents/icaldownload.py

@@ -2,11 +2,15 @@
 Downloads an iCal url or reads an iCal file.
 """
 
+from __future__ import annotations
+
+from contextlib import suppress
+from pathlib import Path
+
 import urllib3
-import logging
 
 
-def apple_data_fix(content):
+def apple_data_fix(content: str) -> str:
     """
     Fix Apple tzdata bug.
 
@@ -16,7 +20,7 @@ def apple_data_fix(content):
     return content.replace("TZOFFSETFROM:+5328", "TZOFFSETFROM:+0053")
 
 
-def apple_url_fix(url):
+def apple_url_fix(url: str) -> str:
     """
     Fix Apple URL.
 
@@ -33,17 +37,14 @@ class ICalDownload:
     Downloads or reads and decodes iCal sources.
     """
 
-    def __init__(self, http=None):
-        # Get logger
-        logger = logging.getLogger()
-
+    def __init__(self, http: urllib3.PoolManager | None = None) -> None:
         # default http connection to use
         if http is None:
             http = urllib3.PoolManager()
 
         self.http = http
 
-    def data_from_url(self, url, apple_fix=False):
+    def data_from_url(self, url: str, apple_fix: bool = False) -> str:
         """
         Download iCal data from URL.
 
@@ -59,16 +60,14 @@ def data_from_url(self, url, apple_fix=False):
         if not response.data:
             raise ConnectionError("Could not get data from %s!" % url)
 
-        content_type = response.headers.get("content-type")
-
-        try:
-            encoding = content_type.split("charset=")[1]
-        except (AttributeError, IndexError):
-            encoding = "utf-8"
+        encoding = "utf-8"
+        if content_type := response.headers.get("content-type"):
+            with suppress(AttributeError, IndexError):
+                encoding = content_type.split("charset=")[1]
 
         return self.decode(response.data, encoding, apple_fix=apple_fix)
 
-    def data_from_file(self, file, apple_fix=False):
+    def data_from_file(self, file: str | Path, apple_fix: bool = False) -> str:
         """
         Read iCal data from file.
 
@@ -84,14 +83,18 @@ def data_from_file(self, file, apple_fix=False):
 
         return self.decode(content, apple_fix=apple_fix)
 
-    def data_from_string(self, string_content, apple_fix=False):
+    def data_from_string(
+        self, string_content: bytes | str, apple_fix: bool = False
+    ) -> str:
         if not string_content:
             raise OSError("String content is not readable or is empty!")
 
         return self.decode(string_content, apple_fix=apple_fix)
 
     @staticmethod
-    def decode(content, encoding="utf-8", apple_fix=False):
+    def decode(
+        content: bytes | str, encoding: str = "utf-8", apple_fix: bool = False
+    ) -> str:
         """
         Decode content using the set charset.
 

icalevents/icalevents.py

@@ -1,28 +1,34 @@
+from __future__ import annotations
+
+from datetime import datetime, tzinfo as _tzinfo
+from pathlib import Path
 from threading import Lock, Thread
 
-from .icalparser import parse_events, Event
+import urllib3
+
+from .icalparser import Event, parse_events
 from .icaldownload import ICalDownload
 
 
 # Lock for event data
 event_lock = Lock()
 # Event data
-event_store = {}
+event_store: dict[str, list[Event]] = {}
 # Threads
-threads = {}
+threads: dict[str, list[Thread]] = {}
 
 
 def events(
-    url=None,
-    file=None,
-    string_content=None,
-    start=None,
-    end=None,
-    fix_apple=False,
-    http=None,
-    tzinfo=None,
-    sort=None,
-    strict=False,
+    url: str | None = None,
+    file: str | Path | None = None,
+    string_content: bytes | str | None = None,
+    start: datetime | None = None,
+    end: datetime | None = None,
+    fix_apple: bool = False,
+    http: urllib3.PoolManager | None = None,
+    tzinfo: _tzinfo | None = None,
+    sort: bool = False,
+    strict: bool = False,
 ) -> list[Event]:
     """
     Get all events form the given iCal URL occurring in the given time range.
@@ -42,7 +48,7 @@ def events(
     """
     found_events = []
 
-    content = None
+    content = ""
     ical_download = ICalDownload(http=http)
 
     if url:
@@ -58,13 +64,21 @@ def events(
         content, start=start, end=end, tzinfo=tzinfo, sort=sort, strict=strict
     )
 
-    if sort is True:
+    if sort:
         found_events.sort()
 
     return found_events
 
 
-def request_data(key, url, file, string_content, start, end, fix_apple):
+def request_data(
+    key: str,
+    url: str | None,
+    file: str | Path | None,
+    string_content: bytes | str | None,
+    start: datetime | None,
+    end: datetime | None,
+    fix_apple: bool,
+) -> None:
     """
     Request data, update local data cache and remove this Thread from queue.
 
@@ -93,8 +107,14 @@ def request_data(key, url, file, string_content, start, end, fix_apple):
 
 
 def events_async(
-    key, url=None, file=None, start=None, string_content=None, end=None, fix_apple=False
-):
+    key: str,
+    url: str | None = None,
+    file: str | Path | None = None,
+    start: datetime | None = None,
+    string_content: bytes | str | None = None,
+    end: datetime | None = None,
+    fix_apple: bool = False,
+) -> None:
     """
     Trigger an asynchronous data request.
 
@@ -121,7 +141,7 @@ def events_async(
             threads[key][0].start()
 
 
-def request_finished(key):
+def request_finished(key: str) -> None:
     """
     Remove finished Thread from queue.
 
@@ -134,7 +154,7 @@ def request_finished(key):
             threads[key][0].run()
 
 
-def update_events(key, data):
+def update_events(key: str, data: list[Event]) -> None:
     """
     Set the latest events for a key.
 
@@ -145,7 +165,7 @@ def update_events(key, data):
         event_store[key] = data
 
 
-def latest_events(key):
+def latest_events(key: str) -> list[Event]:
     """
     Get the latest downloaded events for the given key.
 
@@ -158,7 +178,7 @@ def latest_events(key):
     return res
 
 
-def all_done(key):
+def all_done(key: str) -> bool:
     """
     Check if requests for the given key are active.