Enable Ruff

.github/workflows/tests.yml

@@ -11,8 +11,6 @@ permissions:
 
 jobs:
   ruff:
-    # disabled until after a repo-wide ruff pass
-    if: ${{ false }}
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v7

pyproject.toml

@@ -88,5 +88,5 @@ required-version = "==0.16.1"
 target-version = "py312"
 
 [tool.ruff.lint]
-# Baseline defaults from #56; CI scopes UP diagnostics to changed lines.
+# Baseline checks and Python modernization rules, enforced repository-wide.
 select = ["E4", "E7", "E9", "F", "UP"]

src/livepeer_gateway/channel_reader.py

@@ -267,7 +267,7 @@ def __call__(
         """
         url = self.events_url
 
-        async def _read_all(segment: "SegmentReader", *, chunk_size: int = 33 * 1024) -> bytes:
+        async def _read_all(segment: SegmentReader, *, chunk_size: int = 33 * 1024) -> bytes:
             parts = []
             try:
                 reader = segment.make_reader()

src/livepeer_gateway/channel_writer.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 import logging
-from typing import Any, Optional, Protocol
+from typing import Any, Protocol
 
 from .trickle_publisher import SegmentWriter, TricklePublisher
 
@@ -31,12 +31,12 @@ def __init__(
         self.url = url
         self._publisher = TricklePublisher(url, mime_type)
         self._segment_interval = segment_interval
-        self._writer: Optional[SegmentWriter] = None
+        self._writer: SegmentWriter | None = None
         self._lock = asyncio.Lock()
-        self._rotation_task: Optional[asyncio.Task[None]] = None
+        self._rotation_task: asyncio.Task[None] | None = None
         self.start_rotation()
 
-    def start_rotation(self) -> Optional[asyncio.Task[None]]:
+    def start_rotation(self) -> asyncio.Task[None] | None:
         """
         Start periodic segment rotation if running in an event loop.
 

src/livepeer_gateway/control.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 import logging
-from typing import Any, Optional
+from typing import Any
 
 from .trickle_publisher import (
     TricklePublisher,
@@ -23,7 +23,7 @@ def __init__(
         self.control_url = control_url
         self._publisher = TricklePublisher(control_url, mime_type)
         self._keepalive_interval_s = 10.0
-        self._keepalive_task: Optional[asyncio.Task[None]] = None
+        self._keepalive_task: asyncio.Task[None] | None = None
 
     async def write(self, msg: dict[str, Any]) -> None:
         """
@@ -58,7 +58,7 @@ async def close(self) -> None:
                 _LOG.exception("Control keepalive task failed during shutdown")
         await self._publisher.close()
 
-    def start_keepalive(self) -> Optional[asyncio.Task[None]]:
+    def start_keepalive(self) -> asyncio.Task[None] | None:
         """
         Start periodic keepalive messages.
 

src/livepeer_gateway/media_decode.py

@@ -5,7 +5,6 @@
 import time
 from dataclasses import dataclass
 from fractions import Fraction
-from typing import Optional, Union
 
 import av
 
@@ -14,11 +13,11 @@
 class DecodedMediaFrame:
     kind: str
     stream_index: int
-    frame: Union[av.VideoFrame, av.AudioFrame]
-    pts: Optional[int]
-    time_base: Optional[Fraction]
+    frame: av.VideoFrame | av.AudioFrame
+    pts: int | None
+    time_base: Fraction | None
     # pts_time: presentation timestamp in seconds (pts * time_base). Not wall-clock time.
-    pts_time: Optional[float]
+    pts_time: float | None
     demuxed_at: float
     decoded_at: float
 
@@ -27,27 +26,27 @@ class DecodedMediaFrame:
 class VideoDecodedMediaFrame(DecodedMediaFrame):
     width: int
     height: int
-    pix_fmt: Optional[str]
+    pix_fmt: str | None
 
 
 @dataclass(frozen=True)
 class AudioDecodedMediaFrame(DecodedMediaFrame):
-    sample_rate: Optional[int]
-    layout: Optional[str]
-    format: Optional[str]
-    samples: Optional[int]
+    sample_rate: int | None
+    layout: str | None
+    format: str | None
+    samples: int | None
 
 
 @dataclass(frozen=True)
 class DemuxedMediaPacket:
     kind: str
     stream_index: int
     packet: av.Packet
-    pts: Optional[int]
-    dts: Optional[int]
-    time_base: Optional[Fraction]
-    pts_time: Optional[float]
-    dts_time: Optional[float]
+    pts: int | None
+    dts: int | None
+    time_base: Fraction | None
+    pts_time: float | None
+    dts_time: float | None
     is_keyframe: bool
     size: int
     demuxed_at: float
@@ -120,7 +119,7 @@ class _BlockingByteStream:
     def __init__(self) -> None:
         # TODO: add backpressure (bounded queue + blocking feed) to avoid unbounded
         # memory growth if the decoder thread can't keep up with the producer.
-        self._queue: "queue.Queue[object]" = queue.Queue()
+        self._queue: queue.Queue[object] = queue.Queue()
         self._buffer = bytearray()
         self._closed = False
         self._total_chunks_fed = 0
@@ -207,7 +206,7 @@ def __init__(self, error: BaseException) -> None:
         self.error = error
 
 
-def _fraction_from_time_base(time_base: object) -> Optional[Fraction]:
+def _fraction_from_time_base(time_base: object) -> Fraction | None:
     numerator = getattr(time_base, "numerator", None)
     denominator = getattr(time_base, "denominator", None)
     if numerator is not None and denominator is not None:
@@ -221,7 +220,7 @@ def _fraction_from_time_base(time_base: object) -> Optional[Fraction]:
         return None
 
 
-def _time_from_pts(pts: Optional[int], time_base: Optional[Fraction]) -> Optional[float]:
+def _time_from_pts(pts: int | None, time_base: Fraction | None) -> float | None:
     if pts is None or time_base is None:
         return None
     try:
@@ -231,7 +230,7 @@ def _time_from_pts(pts: Optional[int], time_base: Optional[Fraction]) -> Optiona
 
 
 def _build_decoded_frame(
-    frame: Union[av.VideoFrame, av.AudioFrame],
+    frame: av.VideoFrame | av.AudioFrame,
     *,
     stream_index: int,
     demuxed_at: float,
@@ -280,7 +279,7 @@ def _stream_kind(packet: av.Packet) -> str:
     return "unknown"
 
 
-def _packet_time_base(packet: av.Packet) -> Optional[Fraction]:
+def _packet_time_base(packet: av.Packet) -> Fraction | None:
     time_base = getattr(packet, "time_base", None)
     if time_base is None:
         stream = getattr(packet, "stream", None)
@@ -313,7 +312,7 @@ def _build_demuxed_packet(
     )
 
 
-def _item_pts_time(item: object) -> Optional[float]:
+def _item_pts_time(item: object) -> float | None:
     if isinstance(item, (DecodedMediaFrame, DemuxedMediaPacket)):
         return item.pts_time
     return None
@@ -323,16 +322,16 @@ class _MpegTsOutputWorker:
     def __init__(self, *, thread_name: str) -> None:
         self._reader = _BlockingByteStream()
         # TODO: add backpressure here too (bounded queue) if callers are slow to consume.
-        self._output: "queue.Queue[object]" = queue.Queue()
+        self._output: queue.Queue[object] = queue.Queue()
         self._total_output_items_enqueued = 0
         self._total_output_items_dequeued = 0
         self._output_wait_s = 0.0
         # Aggregate source-PTS media-time watermarks across all streams for
         # output-queue telemetry. Best-effort snapshots updated from producer
         # (decoder thread) and consumer (get()) without per-op locks.
-        self._max_pts_time_enqueued: Optional[float] = None
-        self._max_pts_time_dequeued: Optional[float] = None
-        self._min_pts_time_dequeued: Optional[float] = None
+        self._max_pts_time_enqueued: float | None = None
+        self._max_pts_time_dequeued: float | None = None
+        self._min_pts_time_dequeued: float | None = None
         self._stop = threading.Event()
         self._thread = threading.Thread(target=self._run, name=thread_name, daemon=True)
 
@@ -419,7 +418,7 @@ def __init__(self) -> None:
         super().__init__(thread_name="MpegTsDecoder")
 
     def _run(self) -> None:
-        container: Optional[av.container.InputContainer] = None
+        container: av.container.InputContainer | None = None
         try:
             container = av.open(self._reader, format="mpegts", mode="r")
             for packet in container.demux():
@@ -459,7 +458,7 @@ def __init__(self) -> None:
         super().__init__(thread_name="MpegTsPacketDemuxer")
 
     def _run(self) -> None:
-        container: Optional[av.container.InputContainer] = None
+        container: av.container.InputContainer | None = None
         try:
             container = av.open(self._reader, format="mpegts", mode="r")
             for packet in container.demux():
@@ -488,7 +487,7 @@ def is_decoder_end(item: object) -> bool:
     return item is _END
 
 
-def decoder_error(item: object) -> Optional[BaseException]:
+def decoder_error(item: object) -> BaseException | None:
     if isinstance(item, _DecoderError):
         return item.error
     return None

src/livepeer_gateway/media_output.py

@@ -12,7 +12,6 @@
 from enum import Enum
 from contextlib import suppress
 from collections.abc import AsyncIterator, Awaitable, Callable, Collection
-from typing import Optional
 
 from .errors import LivepeerGatewayError
 from .media_decode import (
@@ -71,8 +70,8 @@ class MediaOutputStats:
     # decoder: decode pipeline queue metrics, including the distinction between
     # queued_bytes in the cross-thread input queue and buffered_bytes already
     # staged in the internal bytearray used to satisfy read() calls.
-    decoder: Optional[DecoderQueueStats]
-    subscriber: Optional[TrickleSubscriberStats]
+    decoder: DecoderQueueStats | None
+    subscriber: TrickleSubscriberStats | None
 
     def __str__(self) -> str:
         return (
@@ -140,7 +139,7 @@ def __init__(
         *,
         start_seq: int = -2,
         max_retries: int = 5,
-        max_segment_bytes: Optional[int] = None,
+        max_segment_bytes: int | None = None,
         connection_close: bool = False,
         chunk_size: int = 64 * 1024,
         max_segments: int = 5,
@@ -165,15 +164,15 @@ def __init__(
         self.on_frame = on_frame
         self.on_packet = on_packet
 
-        self._sub: Optional[TrickleSubscriber] = None
+        self._sub: TrickleSubscriber | None = None
         self._segments: list[SegmentReader] = []
         self._lock = asyncio.Lock()
         self._eos = False
         self._next_local_seq = 0
         self._base_seq = 0
         self._started_at = time.time()
-        self._processor: Optional[MpegTsDecoder | MpegTsPacketDemuxer] = None
-        self._last_decoder_stats: Optional[DecoderQueueStats] = None
+        self._processor: MpegTsDecoder | MpegTsPacketDemuxer | None = None
+        self._last_decoder_stats: DecoderQueueStats | None = None
         self._bytes_callback_task: asyncio.Task[None] | None = None
         self._frame_callback_task: asyncio.Task[None] | None = None
         self._packet_callback_task: asyncio.Task[None] | None = None
@@ -481,7 +480,7 @@ async def _iter_bytes(
     async def _next_segment(
         self,
         seq: int,
-    ) -> Optional[SegmentReader]:
+    ) -> SegmentReader | None:
         """
         Return the segment at seq, lazily advancing the subscriber if needed.
         """
@@ -571,7 +570,7 @@ async def close(self, *, wait_callbacks: bool = True, timeout: float | None = 10
         if self._callback_errors:
             raise self._callback_errors[0]
 
-    async def __aenter__(self) -> "MediaOutput":
+    async def __aenter__(self) -> MediaOutput:
         self.start_callbacks()
         return self
 
@@ -605,7 +604,7 @@ def get_stats(self) -> MediaOutputStats:
         )
 
 
-def _normalize_content_type(value: Optional[str]) -> Optional[str]:
+def _normalize_content_type(value: str | None) -> str | None:
     if value is None:
         return None
     return value.split(";", 1)[0].strip().lower()
@@ -624,7 +623,7 @@ def _normalize_accepted_content_types(
     return normalized
 
 
-def _require_content_type(value: Optional[str], accepted: frozenset[str]) -> None:
+def _require_content_type(value: str | None, accepted: frozenset[str]) -> None:
     normalized = _normalize_content_type(value)
     if normalized not in accepted:
         raise LivepeerGatewayError(