Strict type checking fixes

livekit-api/livekit/api/access_token.py

@@ -281,9 +281,9 @@ def verify(self, token: str, *, verify_signature: bool = True) -> Claims:
         return grant_claims
 
 
-def camel_to_snake(t: str):
+def camel_to_snake(t: str) -> str:
     return re.sub(r"(?<!^)(?=[A-Z])", "_", t).lower()
 
 
-def snake_to_lower_camel(t: str):
+def snake_to_lower_camel(t: str) -> str:
     return "".join(word.capitalize() if i else word for i, word in enumerate(t.split("_")))

livekit-api/livekit/api/livekit_api.py

@@ -6,7 +6,7 @@
 from .sip_service import SipService
 from .agent_dispatch_service import AgentDispatchService
 from .connector_service import ConnectorService
-from typing import Optional
+from typing import Any, Optional
 
 
 class LiveKitAPI:
@@ -96,21 +96,21 @@ def connector(self) -> ConnectorService:
         """Instance of the ConnectorService"""
         return self._connector
 
-    async def aclose(self):
+    async def aclose(self) -> None:
         """Close the API client
 
         Call this before your application exits or when the API client is no longer needed."""
         # we do not close custom sessions, that's up to the caller
-        if not self._custom_session:
+        if not self._custom_session and self._session is not None:
             await self._session.close()
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> "LiveKitAPI":
         """@private
 
         Support for `async with`"""
         return self
 
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         """@private
 
         Support for `async with`"""

livekit-rtc/livekit/rtc/_ffi_client.py

@@ -34,7 +34,7 @@
 atexit.register(_resource_files.close)
 
 
-def _lib_name():
+def _lib_name() -> Optional[str]:
     if platform.system() == "Linux":
         return "liblivekit_ffi.so"
     elif platform.system() == "Darwin":
@@ -44,7 +44,7 @@ def _lib_name():
     return None
 
 
-def get_ffi_lib():
+def get_ffi_lib() -> ctypes.CDLL:
     # allow to override the lib path using an env var
     libpath = os.environ.get("LIVEKIT_LIB_PATH", "").strip()
     if libpath:
@@ -73,7 +73,7 @@ def __init__(self, handle: int) -> None:
         self.handle = handle
         self._disposed = False
 
-    def __del__(self):
+    def __del__(self) -> None:
         self.dispose()
 
     @property
@@ -149,7 +149,7 @@ def unsubscribe(self, queue: Queue[T]) -> None:
                     break
 
 
-@ctypes.CFUNCTYPE(None, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t)
+@ctypes.CFUNCTYPE(None, ctypes.POINTER(ctypes.c_uint8), ctypes.c_size_t)  # type: ignore[untyped-decorator]
 def ffi_event_callback(
     data_ptr: ctypes.POINTER(ctypes.c_uint8),  # type: ignore
     data_len: ctypes.c_size_t,
@@ -255,7 +255,7 @@ def __init__(self) -> None:
         ffi_lib = self._ffi_lib
 
         @atexit.register
-        def _dispose_lk_ffi():
+        def _dispose_lk_ffi() -> None:
             ffi_lib.livekit_ffi_dispose()
 
     @property

livekit-rtc/livekit/rtc/_utils.py

@@ -17,16 +17,16 @@
 from collections import deque
 import ctypes
 import random
-from typing import Callable, Generator, Generic, List, TypeVar, Union
+from typing import Any, Callable, Generator, Generic, List, TypeVar, Union
 
 logger = logging.getLogger("livekit")
 
 
 class classproperty(object):
-    def __init__(self, f):
-        self.f = classmethod(f)
+    def __init__(self, f: Callable[..., Any]) -> None:
+        self.f: Any = classmethod(f)
 
-    def __get__(self, *a):
+    def __get__(self, *a: Any) -> Any:
         return self.f.__get__(*a)()
 
 
@@ -146,7 +146,7 @@ async def join(self) -> None:
 _base62_characters = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
 
 
-def generate_random_base62(length=12):
+def generate_random_base62(length: int = 12) -> str:
     """
     Generate a random base62 encoded string of a specified length.
 

livekit-rtc/livekit/rtc/audio_frame.py

@@ -193,7 +193,7 @@ def __repr__(self) -> str:
         )
 
     @classmethod
-    def __get_pydantic_core_schema__(cls, *_: Any):
+    def __get_pydantic_core_schema__(cls, *_: Any) -> Any:
         from pydantic_core import core_schema
         import base64
 

livekit-rtc/livekit/rtc/audio_stream.py

@@ -65,7 +65,7 @@ def __init__(
         num_channels: int = 1,
         frame_size_ms: int | None = None,
         noise_cancellation: Optional[NoiseCancellationOptions | FrameProcessor[AudioFrame]] = None,
-        **kwargs,
+        **kwargs: Any,
     ) -> None:
         """Initialize an `AudioStream` instance.
 
@@ -273,7 +273,7 @@ def _create_owned_stream_from_participant(
         resp = FfiClient.instance.request(req)
         return resp.audio_stream_from_participant.stream
 
-    async def _run(self):
+    async def _run(self) -> None:
         while True:
             event = await self._ffi_queue.wait_for(self._is_event)
             audio_event: proto_audio_frame.AudioStreamEvent = event.audio_stream_event

livekit-rtc/livekit/rtc/data_stream.py

@@ -66,10 +66,10 @@ def __init__(
         )
         self._queue: asyncio.Queue[proto_DataStream.Chunk | None] = asyncio.Queue()
 
-    async def _on_chunk_update(self, chunk: proto_DataStream.Chunk):
+    async def _on_chunk_update(self, chunk: proto_DataStream.Chunk) -> None:
         await self._queue.put(chunk)
 
-    async def _on_stream_close(self, trailer: proto_DataStream.Trailer):
+    async def _on_stream_close(self, trailer: proto_DataStream.Trailer) -> None:
         self.info.attributes = self.info.attributes or {}
         self.info.attributes.update(trailer.attributes)
         await self._queue.put(None)
@@ -114,10 +114,10 @@ def __init__(self, header: proto_DataStream.Header, capacity: int = 0) -> None:
         )
         self._queue: asyncio.Queue[proto_DataStream.Chunk | None] = asyncio.Queue(capacity)
 
-    async def _on_chunk_update(self, chunk: proto_DataStream.Chunk):
+    async def _on_chunk_update(self, chunk: proto_DataStream.Chunk) -> None:
         await self._queue.put(chunk)
 
-    async def _on_stream_close(self, trailer: proto_DataStream.Trailer):
+    async def _on_stream_close(self, trailer: proto_DataStream.Trailer) -> None:
         self.info.attributes = self.info.attributes or {}
         self.info.attributes.update(trailer.attributes)
         await self._queue.put(None)
@@ -166,7 +166,7 @@ def __init__(
         self._sender_identity = sender_identity or self._local_participant.identity
         self._closed = False
 
-    async def _send_header(self):
+    async def _send_header(self) -> None:
         req = proto_ffi.FfiRequest(
             send_stream_header=proto_room.SendStreamHeaderRequest(
                 header=self._header,
@@ -188,7 +188,7 @@ async def _send_header(self):
         if cb.send_stream_header.error:
             raise ConnectionError(cb.send_stream_header.error)
 
-    async def _send_chunk(self, chunk: proto_DataStream.Chunk):
+    async def _send_chunk(self, chunk: proto_DataStream.Chunk) -> None:
         if self._closed:
             raise RuntimeError(f"Cannot send chunk after stream is closed: {chunk}")
         req = proto_ffi.FfiRequest(
@@ -212,7 +212,7 @@ async def _send_chunk(self, chunk: proto_DataStream.Chunk):
         if cb.send_stream_chunk.error:
             raise ConnectionError(cb.send_stream_chunk.error)
 
-    async def _send_trailer(self, trailer: proto_DataStream.Trailer):
+    async def _send_trailer(self, trailer: proto_DataStream.Trailer) -> None:
         req = proto_ffi.FfiRequest(
             send_stream_trailer=proto_room.SendStreamTrailerRequest(
                 trailer=trailer,
@@ -230,10 +230,12 @@ async def _send_trailer(self, trailer: proto_DataStream.Trailer):
         finally:
             FfiClient.instance.queue.unsubscribe(queue)
 
-        if cb.send_stream_chunk.error:
+        if cb.send_stream_trailer.error:
             raise ConnectionError(cb.send_stream_trailer.error)
 
-    async def aclose(self, *, reason: str = "", attributes: Optional[Dict[str, str]] = None):
+    async def aclose(
+        self, *, reason: str = "", attributes: Optional[Dict[str, str]] = None
+    ) -> None:
         if self._closed:
             raise RuntimeError("Stream already closed")
         self._closed = True
@@ -281,7 +283,7 @@ def __init__(
         )
         self._write_lock = asyncio.Lock()
 
-    async def write(self, text: str):
+    async def write(self, text: str) -> None:
         async with self._write_lock:
             for chunk in split_utf8(text, STREAM_CHUNK_SIZE):
                 content = chunk
@@ -333,7 +335,7 @@ def __init__(
         )
         self._write_lock = asyncio.Lock()
 
-    async def write(self, data: bytes):
+    async def write(self, data: bytes) -> None:
         async with self._write_lock:
             chunked_data = [
                 data[i : i + STREAM_CHUNK_SIZE] for i in range(0, len(data), STREAM_CHUNK_SIZE)

livekit-rtc/livekit/rtc/data_track.py

@@ -117,7 +117,7 @@ def is_published(self) -> bool:
         req.local_data_track_is_published.track_handle = self._ffi_handle.handle
 
         resp = FfiClient.instance.request(req)
-        return resp.local_data_track_is_published.is_published
+        return bool(resp.local_data_track_is_published.is_published)
 
     async def unpublish(self) -> None:
         """Unpublishes the track."""
@@ -181,7 +181,7 @@ def is_published(self) -> bool:
         req.remote_data_track_is_published.track_handle = self._ffi_handle.handle
 
         resp = FfiClient.instance.request(req)
-        return resp.remote_data_track_is_published.is_published
+        return bool(resp.remote_data_track_is_published.is_published)
 
     def __repr__(self) -> str:
         return (

livekit-rtc/livekit/rtc/e2ee.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from dataclasses import dataclass, field
-from typing import List, Optional
+from typing import List, Optional, cast
 
 from ._ffi_client import FfiClient
 from ._proto import e2ee_pb2 as proto_e2ee
@@ -89,7 +89,7 @@ def export_shared_key(self, key_index: int) -> bytes:
         req.e2ee.get_shared_key.key_index = key_index
         resp = FfiClient.instance.request(req)
         key = resp.e2ee.get_shared_key.key
-        return key
+        return cast(bytes, key)
 
     def ratchet_shared_key(self, key_index: int) -> bytes:
         """Ratchets the shared encryption key to a new key.
@@ -112,7 +112,7 @@ def ratchet_shared_key(self, key_index: int) -> bytes:
         resp = FfiClient.instance.request(req)
 
         new_key = resp.e2ee.ratchet_shared_key.new_key
-        return new_key
+        return cast(bytes, new_key)
 
     def set_key(self, participant_identity: str, key: bytes, key_index: int) -> None:
         """Sets the encryption key for a specific participant.
@@ -157,7 +157,7 @@ def export_key(self, participant_identity: str, key_index: int) -> bytes:
         req.e2ee.get_key.key_index = key_index
         resp = FfiClient.instance.request(req)
         key = resp.e2ee.get_key.key
-        return key
+        return cast(bytes, key)
 
     def ratchet_key(self, participant_identity: str, key_index: int) -> bytes:
         """Ratchets the encryption key for a specific participant to a new key.
@@ -181,7 +181,7 @@ def ratchet_key(self, participant_identity: str, key_index: int) -> bytes:
 
         resp = FfiClient.instance.request(req)
         new_key = resp.e2ee.ratchet_key.new_key
-        return new_key
+        return cast(bytes, new_key)
 
 
 class FrameCryptor:

livekit-rtc/livekit/rtc/event_emitter.py

@@ -1,6 +1,6 @@
 import inspect
 import asyncio
-from typing import Callable, Dict, Set, Optional, Generic, TypeVar
+from typing import Any, Callable, Dict, Set, Optional, Generic, TypeVar
 
 from .log import logger
 
@@ -14,7 +14,7 @@ def __init__(self) -> None:
         """
         self._events: Dict[T_contra, Set[Callable]] = dict()
 
-    def emit(self, event: T_contra, *args) -> None:
+    def emit(self, event: T_contra, *args: Any) -> None:
         """
         Trigger all callbacks associated with the given event.
 
@@ -104,7 +104,7 @@ def greet_once(name):
         """
         if callback is not None:
 
-            def once_callback(*args, **kwargs):
+            def once_callback(*args: Any, **kwargs: Any) -> None:
                 self.off(event, once_callback)
                 callback(*args, **kwargs)
 

livekit-rtc/livekit/rtc/jupyter.py

@@ -50,7 +50,7 @@ def room_html(url: str, token: str, *, width: str, height: str) -> HTML:
             f'srcdoc="{escaped_content}"></iframe>'
         )
 
-    return HTML(html_text)
+    return HTML(html_text)  # type: ignore[no-untyped-call]
 
 
 def display_room(url: str, token: str, *, width: str = "100%", height: str = "110px") -> None:
@@ -65,4 +65,4 @@ def display_room(url: str, token: str, *, width: str = "100%", height: str = "11
         The rendered HTML will include the provided `url` and `token` in plain text.
         Avoid using sensitive tokens in public notebooks (e.g., tokens with long expiration times).
     """
-    display(room_html(url, token, width=width, height=height))
+    display(room_html(url, token, width=width, height=height))  # type: ignore[no-untyped-call]