Fix linting errors

Signed-off-by: Jared O'Connell <[email protected]>

Author: jaredoconnell

src/guidellm/utils/__init__.py

@@ -81,7 +81,6 @@
     "EndlessTextCreator",
     "InfoMixin",
     "IntegerRangeSampler",
-    "camelize_str",
     "InterProcessMessaging",
     "InterProcessMessagingManagerQueue",
     "InterProcessMessagingPipe",
@@ -107,14 +106,15 @@
     "ThreadSafeSingletonMixin",
     "TimeRunningStats",
     "all_defined",
+    "camelize_str",
     "check_load_processor",
     "clean_text",
     "filter_text",
     "format_value_display",
     "get_literal_vals",
     "is_punctuation",
-    "recursive_key_update",
     "load_text",
+    "recursive_key_update",
     "safe_add",
     "safe_divide",
     "safe_format_timestamp",

src/guidellm/utils/encoding.py

@@ -12,10 +12,10 @@
 
 import json
 from collections.abc import Mapping
-from typing import Annotated, Any, cast, ClassVar, Generic, Literal, Optional, TypeVar
+from typing import Annotated, Any, ClassVar, Generic, Literal, Optional, TypeVar, cast
 
 try:
-    import msgpack # type: ignore[import-untyped] # Optional dependency
+    import msgpack  # type: ignore[import-untyped] # Optional dependency
     from msgpack import Packer, Unpacker
 
     HAS_MSGPACK = True
@@ -24,16 +24,20 @@
     HAS_MSGPACK = False
 
 try:
-    from msgspec.msgpack import Decoder as MsgspecDecoder # type: ignore[import-not-found] # Optional dependency
-    from msgspec.msgpack import Encoder as MsgspecEncoder # type: ignore[import-not-found] # Optional dependency
+    from msgspec.msgpack import (  # type: ignore[import-not-found] # Optional dependency
+        Decoder as MsgspecDecoder,
+    )
+    from msgspec.msgpack import (  # type: ignore[import-not-found] # Optional dependency
+        Encoder as MsgspecEncoder,
+    )
 
     HAS_MSGSPEC = True
 except ImportError:
     MsgspecDecoder = MsgspecEncoder = None
     HAS_MSGSPEC = False
 
 try:
-    import orjson # type: ignore[import-not-found] # Optional dependency
+    import orjson  # type: ignore[import-not-found] # Optional dependency
 
     HAS_ORJSON = True
 except ImportError:
@@ -116,7 +120,7 @@ def encode_message(
         """
         serialized = serializer.serialize(obj) if serializer else obj
 
-        return cast(MsgT, encoder.encode(serialized) if encoder else serialized)
+        return cast("MsgT", encoder.encode(serialized) if encoder else serialized)
 
     @classmethod
     def decode_message(
@@ -137,7 +141,9 @@ def decode_message(
         """
         serialized = encoder.decode(message) if encoder else message
 
-        return cast(ObjT, serializer.deserialize(serialized) if serializer else serialized)
+        return cast(
+            "ObjT", serializer.deserialize(serialized) if serializer else serialized
+        )
 
     def __init__(
         self,
@@ -296,7 +302,14 @@ def _get_available_encoder_decoder(
         return None, None, None
 
 
-PayloadType = Literal['pydantic', 'python', 'collection_tuple', 'collection_sequence', 'collection_mapping']
+PayloadType = Literal[
+    "pydantic",
+    "python",
+    "collection_tuple",
+    "collection_sequence",
+    "collection_mapping",
+]
+
 
 class Serializer:
     """
@@ -518,7 +531,9 @@ def to_sequence(self, obj: Any) -> str | Any:
             payload_type = "python"
             payload = self.to_sequence_python(obj)
 
-        return self.pack_next_sequence(payload_type, payload if payload is not None else "", None)
+        return self.pack_next_sequence(
+            payload_type, payload if payload is not None else "", None
+        )
 
     def from_sequence(self, data: str | Any) -> Any:  # noqa: C901, PLR0912
         """
@@ -709,8 +724,10 @@ def pack_next_sequence(  # noqa: C901, PLR0912
             delimiter = "|"
 
         # Type ignores because types are enforced at runtime
-        next_sequence = payload_type + delimiter + payload_len_output + delimiter + payload # type: ignore[operator]
-        return current + next_sequence if current else next_sequence # type: ignore[operator]
+        next_sequence = (
+            payload_type + delimiter + payload_len_output + delimiter + payload  # type: ignore[operator]
+        )
+        return current + next_sequence if current else next_sequence  # type: ignore[operator]
 
     def unpack_next_sequence(  # noqa: C901, PLR0912
         self, data: str | bytes

src/guidellm/utils/messaging.py

@@ -22,7 +22,7 @@
 from multiprocessing.managers import SyncManager
 from multiprocessing.synchronize import Event as ProcessingEvent
 from threading import Event as ThreadingEvent
-from typing import Any, Callable, cast, Generic, List, Protocol, TypeVar
+from typing import Any, Callable, Generic, Protocol, TypeVar, cast
 
 import culsans
 from pydantic import BaseModel
@@ -50,6 +50,7 @@
 
 CheckStopCallableT = Callable[[bool, int], bool]
 
+
 class MessagingStopCallback(Protocol):
     """Protocol for evaluating stop conditions in messaging operations."""
 
@@ -248,14 +249,17 @@ async def stop(self):
         if self.shutdown_event is not None:
             self.shutdown_event.set()
         else:
-            raise RuntimeError("shutdown_event is not set; was start() not called or is this a redundant stop() call?")
+            raise RuntimeError(
+                "shutdown_event is not set; was start() not called or "
+                "is this a redundant stop() call?"
+            )
         tasks = [self.send_task, self.receive_task]
-        tasks_to_run: List[asyncio.Task[Any]] = [task for task in tasks if task is not None]
+        tasks_to_run: list[asyncio.Task[Any]] = [
+            task for task in tasks if task is not None
+        ]
         if len(tasks_to_run) > 0:
             with contextlib.suppress(asyncio.CancelledError):
-                await asyncio.gather(
-                    *tasks_to_run, return_exceptions=True
-                )
+                await asyncio.gather(*tasks_to_run, return_exceptions=True)
         self.send_task = None
         self.receive_task = None
         if self.worker_index is None:
@@ -354,7 +358,9 @@ async def get(self, timeout: float | None = None) -> ReceiveMessageT:
         :return: Decoded message from the receive buffer
         """
         if self.buffer_receive_queue is None:
-            raise RuntimeError("buffer receive queue is None; check start()/stop() calls")
+            raise RuntimeError(
+                "buffer receive queue is None; check start()/stop() calls"
+            )
         return await asyncio.wait_for(
             self.buffer_receive_queue.async_get(), timeout=timeout
         )
@@ -367,7 +373,9 @@ def get_sync(self, timeout: float | None = None) -> ReceiveMessageT:
         :return: Decoded message from the receive buffer
         """
         if self.buffer_receive_queue is None:
-            raise RuntimeError("buffer receive queue is None; check start()/stop() calls")
+            raise RuntimeError(
+                "buffer receive queue is None; check start()/stop() calls"
+            )
         if timeout is not None and timeout <= 0:
             return self.buffer_receive_queue.get_nowait()
         else:
@@ -381,7 +389,9 @@ async def put(self, item: SendMessageT, timeout: float | None = None):
         :param timeout: Maximum time to wait for buffer space
         """
         if self.buffer_send_queue is None:
-            raise RuntimeError("buffer receive queue is None; check start()/stop() calls")
+            raise RuntimeError(
+                "buffer receive queue is None; check start()/stop() calls"
+            )
         await asyncio.wait_for(self.buffer_send_queue.async_put(item), timeout=timeout)
 
     def put_sync(self, item: SendMessageT, timeout: float | None = None):
@@ -392,7 +402,9 @@ def put_sync(self, item: SendMessageT, timeout: float | None = None):
         :param timeout: Maximum time to wait for buffer space, if <=0 uses put_nowait
         """
         if self.buffer_send_queue is None:
-            raise RuntimeError("buffer receive queue is None; check start()/stop() calls")
+            raise RuntimeError(
+                "buffer receive queue is None; check start()/stop() calls"
+            )
         if timeout is not None and timeout <= 0:
             self.buffer_send_queue.put_nowait(item)
         else:
@@ -457,6 +469,7 @@ class InterProcessMessagingQueue(InterProcessMessaging[SendMessageT, ReceiveMess
         # Create worker copy for distributed processing
         worker_messaging = messaging.create_worker_copy(worker_index=0)
     """
+
     pending_queue: multiprocessing.Queue | queue.Queue[Any] | None
     done_queue: multiprocessing.Queue | queue.Queue[Any] | None
 
@@ -545,15 +558,15 @@ async def stop(self):
             with contextlib.suppress(queue.Empty):
                 while True:
                     self.pending_queue.get_nowait()
-            if hasattr(self.pending_queue, 'close'):
+            if hasattr(self.pending_queue, "close"):
                 self.pending_queue.close()
 
             if self.done_queue is None:
                 raise RuntimeError("done_queue is None; was stop() already called?")
             with contextlib.suppress(queue.Empty):
                 while True:
                     self.done_queue.get_nowait()
-            if hasattr(self.done_queue, 'close'):
+            if hasattr(self.done_queue, "close"):
                 self.done_queue.close()
 
         self.pending_queue = None
@@ -618,7 +631,9 @@ def _send_messages_task_thread(  # noqa: C901, PLR0912
                         item = next(send_items_iter)
                     else:
                         if self.buffer_send_queue is None:
-                            raise RuntimeError("buffer_send_queue is None; was stop() already called?")
+                            raise RuntimeError(
+                                "buffer_send_queue is None; was stop() already called?"
+                            )
                         item = self.buffer_send_queue.sync_get(
                             timeout=self.poll_interval
                         )
@@ -632,16 +647,22 @@ def _send_messages_task_thread(  # noqa: C901, PLR0912
                     if self.worker_index is None:
                         # Main publisher
                         if self.pending_queue is None:
-                            raise RuntimeError("pending_queue is None; was stop() already called?")
+                            raise RuntimeError(
+                                "pending_queue is None; was stop() already called?"
+                            )
                         self.pending_queue.put(pending_item, timeout=self.poll_interval)
                     else:
                         # Worker
                         if self.done_queue is None:
-                            raise RuntimeError("done_queue is None; was stop() already called?")
+                            raise RuntimeError(
+                                "done_queue is None; was stop() already called?"
+                            )
                         self.done_queue.put(pending_item, timeout=self.poll_interval)
                     if send_items_iter is None:
                         if self.buffer_send_queue is None:
-                            raise RuntimeError("buffer_send_queue is None; was stop() already called?")
+                            raise RuntimeError(
+                                "buffer_send_queue is None; was stop() already called?"
+                            )
                         self.buffer_send_queue.task_done()
                     pending_item = None
                 except (culsans.QueueFull, queue.Full):
@@ -663,12 +684,16 @@ def _receive_messages_task_thread(  # noqa: C901
                     if self.worker_index is None:
                         # Main publisher
                         if self.done_queue is None:
-                            raise RuntimeError("done_queue is None; check start()/stop() calls")
+                            raise RuntimeError(
+                                "done_queue is None; check start()/stop() calls"
+                            )
                         item = self.done_queue.get(timeout=self.poll_interval)
                     else:
                         # Worker
                         if self.pending_queue is None:
-                            raise RuntimeError("pending_queue is None; check start()/stop() calls")
+                            raise RuntimeError(
+                                "pending_queue is None; check start()/stop() calls"
+                            )
                         item = self.pending_queue.get(timeout=self.poll_interval)
                     pending_item = message_encoding.decode(item)
                     queue_empty_count = 0
@@ -685,8 +710,12 @@ def _receive_messages_task_thread(  # noqa: C901
                         )
 
                     if self.buffer_receive_queue is None:
-                        raise RuntimeError("buffer_receive_queue is None; check start()/stop() calls")
-                    self.buffer_receive_queue.sync_put(cast(ReceiveMessageT, received_item))
+                        raise RuntimeError(
+                            "buffer_receive_queue is None; check start()/stop() calls"
+                        )
+                    self.buffer_receive_queue.sync_put(
+                        cast("ReceiveMessageT", received_item)
+                    )
                     pending_item = None
                     received_item = None
                 except (culsans.QueueFull, queue.Full):
@@ -863,9 +892,7 @@ def __init__(
 
         self.pipes: list[tuple[Connection, Connection]]
         if pipe is None:
-            self.pipes = [
-                self.mp_context.Pipe(duplex=True) for _ in range(num_workers)
-            ]
+            self.pipes = [self.mp_context.Pipe(duplex=True) for _ in range(num_workers)]
         else:
             self.pipes = [pipe]
 
@@ -969,7 +996,7 @@ def create_receive_messages_threads(
                 )
             ]
 
-    def _send_messages_task_thread(  # noqa: C901, PLR0912
+    def _send_messages_task_thread(  # noqa: C901, PLR0912, PLR0915
         self,
         pipe: tuple[Connection, Connection],
         send_items: Iterable[Any] | None,
@@ -1009,7 +1036,9 @@ def _background_pipe_recv():
                             item = next(send_items_iter)
                         else:
                             if self.buffer_send_queue is None:
-                                raise RuntimeError("buffer_send_queue is None; check start()/stop() calls")
+                                raise RuntimeError(
+                                    "buffer_send_queue is None; check start()/stop() calls"  # noqa: E501
+                                )
                             item = self.buffer_send_queue.sync_get(
                                 timeout=self.poll_interval
                             )
@@ -1028,7 +1057,9 @@ def _background_pipe_recv():
                                 pipe_item = pending_item
                         if send_items_iter is None:
                             if self.buffer_send_queue is None:
-                                raise RuntimeError("buffer_send_queue is None; check start()/stop() calls")
+                                raise RuntimeError(
+                                    "buffer_send_queue is None; check start()/stop() calls"  # noqa: E501
+                                )
                             self.buffer_send_queue.task_done()
                         pending_item = None
                     except (culsans.QueueFull, queue.Full):
@@ -1071,8 +1102,12 @@ def _receive_messages_task_thread(  # noqa: C901
                             else receive_callback(pending_item)
                         )
                     if self.buffer_receive_queue is None:
-                        raise RuntimeError("buffer receive queue is None; check start()/stop() calls")
-                    self.buffer_receive_queue.sync_put(cast(ReceiveMessageT, received_item))
+                        raise RuntimeError(
+                            "buffer receive queue is None; check start()/stop() calls"
+                        )
+                    self.buffer_receive_queue.sync_put(
+                        cast("ReceiveMessageT", received_item)
+                    )
                     pending_item = None
                     received_item = None
                 except (culsans.QueueFull, queue.Full):

src/guidellm/utils/pydantic_utils.py

@@ -11,7 +11,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import Any, cast, ClassVar, Generic, TypeVar
+from typing import Any, ClassVar, Generic, TypeVar, cast
 
 from pydantic import BaseModel, ConfigDict, Field, GetCoreSchemaHandler
 from pydantic_core import CoreSchema, core_schema
@@ -300,7 +300,7 @@ def register_decorator(
         super().register_decorator(clazz, name=name)
         cls.reload_schema()
 
-        return cast(RegisterClassT, clazz)
+        return cast("RegisterClassT", clazz)
 
     @classmethod
     def __get_pydantic_core_schema__(

src/guidellm/utils/registry.py

@@ -19,7 +19,9 @@
 
 RegistryObjT = TypeVar("RegistryObjT")
 """Generic type variable for objects managed by the registry system."""
-RegisterT = TypeVar("RegisterT", bound=type)  # Must be bound to type to ensure __name__ is available.
+RegisterT = TypeVar(
+    "RegisterT", bound=type
+)  # Must be bound to type to ensure __name__ is available.
 """Generic type variable for the args and return values within the registry."""