From a6ff11e38377f25900316544f3dd576dc8b94c40 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Fri, 23 Oct 2026 15:02:08 -0700 Subject: [PATCH] Enable Ruff --- .github/workflows/tests.yml | 2 - pyproject.toml | 2 +- src/livepeer_gateway/channel_reader.py | 2 +- src/livepeer_gateway/channel_writer.py | 8 +-- src/livepeer_gateway/control.py | 6 +- src/livepeer_gateway/media_decode.py | 55 ++++++++--------- src/livepeer_gateway/media_output.py | 21 +++---- src/livepeer_gateway/media_publish.py | 71 +++++++++++----------- src/livepeer_gateway/remote_signer.py | 10 +-- src/livepeer_gateway/segment_reader.py | 14 ++--- src/livepeer_gateway/token.py | 4 +- src/livepeer_gateway/trickle_publisher.py | 42 ++++++------- src/livepeer_gateway/trickle_subscriber.py | 19 +++--- 13 files changed, 126 insertions(+), 130 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 66c8076..9e84691 100644 --- a/.github/workflows/tests.yml +++ b/.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 diff --git a/pyproject.toml b/pyproject.toml index c405c6e..6e9a06b 100644 --- a/pyproject.toml +++ b/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"] diff --git a/src/livepeer_gateway/channel_reader.py b/src/livepeer_gateway/channel_reader.py index 1a4694d..63bf1cb 100644 --- a/src/livepeer_gateway/channel_reader.py +++ b/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() diff --git a/src/livepeer_gateway/channel_writer.py b/src/livepeer_gateway/channel_writer.py index c188280..be52675 100644 --- a/src/livepeer_gateway/channel_writer.py +++ b/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. diff --git a/src/livepeer_gateway/control.py b/src/livepeer_gateway/control.py index 4cfc8cf..179e4d6 100644 --- a/src/livepeer_gateway/control.py +++ b/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. diff --git a/src/livepeer_gateway/media_decode.py b/src/livepeer_gateway/media_decode.py index 4413f01..0bd41a8 100644 --- a/src/livepeer_gateway/media_decode.py +++ b/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,15 +26,15 @@ 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) @@ -43,11 +42,11 @@ 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 diff --git a/src/livepeer_gateway/media_output.py b/src/livepeer_gateway/media_output.py index a634304..54bd6ff 100644 --- a/src/livepeer_gateway/media_output.py +++ b/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( diff --git a/src/livepeer_gateway/media_publish.py b/src/livepeer_gateway/media_publish.py index f60de61..1cb8c04 100644 --- a/src/livepeer_gateway/media_publish.py +++ b/src/livepeer_gateway/media_publish.py @@ -9,7 +9,8 @@ import time from dataclasses import dataclass, field from fractions import Fraction -from typing import Optional, Set, Awaitable, Any, BinaryIO +from typing import Any, BinaryIO +from collections.abc import Awaitable import av from av.video.frame import PictureType @@ -47,7 +48,7 @@ def _rescale_pts(pts: int, src_tb: Fraction, dst_tb: Fraction) -> int: return int(round(float((Fraction(pts) * src_tb) / dst_tb))) -def _normalize_fps(fps: Optional[float]) -> int: +def _normalize_fps(fps: float | None) -> int: if fps is None or not math.isfinite(fps) or fps <= 0: fps = 30.0 return max(1, int(round(fps))) @@ -69,7 +70,7 @@ class VideoOutputConfig: # Local queue depth for this track before frames are dropped. queue_size: int = 8 # Target output FPS hint passed to the encoder. - fps: Optional[float] = None + fps: float | None = None # Target GOP/keyframe cadence used for segment boundaries. keyframe_interval_s: float = 2.0 # FFmpeg/PyAV encoder name for this output stream. @@ -96,9 +97,9 @@ class AudioOutputConfig: # FFmpeg/PyAV encoder name for this output stream. codec: str = "libopus" # Target output sample rate. None derives from first frame. - sample_rate: Optional[int] = None + sample_rate: int | None = None # Target output channel layout. None derives from first frame. - layout: Optional[str] = None + layout: str | None = None # Output sample format presented to the encoder. format: str = "flt" @@ -218,12 +219,12 @@ class MediaPublishTrack: def __init__( self, - owner: "MediaPublish", + owner: MediaPublish, *, kind: str, config: VideoOutputConfig | AudioOutputConfig, index: int, - queue: "_FrameQueue", + queue: _FrameQueue, stats: dict[str, int], ) -> None: self._owner = owner @@ -239,14 +240,14 @@ def __init__( self._label: str = "" self._stream: Any = None self._pending_frames: list[av.VideoFrame | av.AudioFrame] = [] - self._first_frame: Optional[av.VideoFrame | av.AudioFrame] = None - self._audio_sample_rate: Optional[int] = None - self._audio_layout: Optional[str] = None + self._first_frame: av.VideoFrame | av.AudioFrame | None = None + self._audio_sample_rate: int | None = None + self._audio_layout: str | None = None self._dropped_timeout: bool = False - self._last_out_pts: Optional[int] = None - self._wallclock_start: Optional[float] = None - self._next_out_pts: Optional[int] = None - self._last_keyframe_time: Optional[float] = None + self._last_out_pts: int | None = None + self._wallclock_start: float | None = None + self._next_out_pts: int | None = None + self._last_keyframe_time: float | None = None self._stopped: bool = False self._audio_resampler: Any = None @@ -290,9 +291,9 @@ def __init__( config.mime_type, ) - self._thread: Optional[threading.Thread] = None - self._loop: Optional[asyncio.AbstractEventLoop] = None - self._segment_tasks: Set[asyncio.Task[None]] = set() + self._thread: threading.Thread | None = None + self._loop: asyncio.AbstractEventLoop | None = None + self._segment_tasks: set[asyncio.Task[None]] = set() self._segment_stream_lock = asyncio.Lock() self._start_lock = threading.Lock() self._track_wait_timeout_s = float(config.track_wait_timeout_s) @@ -306,10 +307,10 @@ def __init__( raise ValueError( "MediaPublishConfig.segment_post_idle_timeout_s must be > 0" ) - self._first_frame_arrived_at: Optional[float] = None + self._first_frame_arrived_at: float | None = None self._closed = False - self._error: Optional[BaseException] = None + self._error: BaseException | None = None self._started_at = time.time() self._stats: dict[str, int] = { "segments_started": 0, @@ -380,9 +381,9 @@ def __init__( self._next_state_index = 0 # Encoder state (owned by the encoder thread). - self._container: Optional[av.container.OutputContainer] = None + self._container: av.container.OutputContainer | None = None self._active_segment: Any = None - self._active_segment_started_at: Optional[float] = None + self._active_segment_started_at: float | None = None # Drain bytes for the rest of the current PyAV segment. self._segment_draining: bool = False @@ -393,7 +394,7 @@ def __init__( def tracks(self) -> tuple[MediaPublishTrack, ...]: return tuple(self._tracks) - def get_tracks(self, kind: Optional[str] = None) -> list[MediaPublishTrack]: + def get_tracks(self, kind: str | None = None) -> list[MediaPublishTrack]: if kind is None: return list(self._tracks) normalized = kind.strip().lower() @@ -599,7 +600,7 @@ def _run_encoder(self) -> None: for track in self._tracks: track._stream = None - def _next_encoder_item(self) -> Optional[tuple[MediaPublishTrack, object]]: + def _next_encoder_item(self) -> tuple[MediaPublishTrack, object] | None: active = [t for t in self._tracks if not t._stopped] if not active: return None @@ -890,8 +891,8 @@ def _flush_encoder(self) -> None: def _compute_pts( self, track: MediaPublishTrack, - pts: Optional[int], - time_base: Optional[Fraction], + pts: int | None, + time_base: Fraction | None, ) -> tuple[float, int]: if pts is not None and time_base is not None: tb = _fraction_from_time_base(time_base) @@ -979,11 +980,11 @@ async def _stream_pipe_to_trickle(self, read_file: BinaryIO) -> None: # Timeout to rotate trickle segments if no bytes from PyAV. idle_timeout_s = self._segment_post_idle_timeout_s - segment_seq: Optional[int] = None + segment_seq: int | None = None # Keep one read in flight across idle timeouts so we eventually # receive late bytes, if any - pending_read: Optional[asyncio.Task[bytes]] = None + pending_read: asyncio.Task[bytes] | None = None # True after we've read payload bytes from this PyAV segment. pipe_past_keyframe = False @@ -1006,7 +1007,7 @@ async def _stream_pipe_to_trickle(self, read_file: BinaryIO) -> None: asyncio.shield(pending_read), timeout=idle_timeout_s, ) - except asyncio.TimeoutError: + except TimeoutError: # NB: This intentionally keeps trickle rolling # accommodate long stalls / inactivity from the # PyAV end, eg during model loading @@ -1181,7 +1182,7 @@ def __init__( self._stats = stats self._debt_skip = debt_skip self._time_debt_s = 0.0 - self._last_encoded_media_time_s: Optional[float] = None + self._last_encoded_media_time_s: float | None = None self._stop_after_current = False self._stop_enqueued = False # Source-PTS media-time watermarks for queue-span telemetry. These are @@ -1189,9 +1190,9 @@ def __init__( # are not locked per op. They are only meaningful for frames that # expose pts + time_base; STOP sentinels and pts-less frames are # skipped. - self._last_put_media_time_s: Optional[float] = None - self._first_get_media_time_s: Optional[float] = None - self._last_get_media_time_s: Optional[float] = None + self._last_put_media_time_s: float | None = None + self._first_get_media_time_s: float | None = None + self._last_get_media_time_s: float | None = None def resize(self, maxsize: int) -> None: if not isinstance(maxsize, int) or isinstance(maxsize, bool): @@ -1235,10 +1236,10 @@ def put(self, item: object) -> None: if item is not _STOP: self._stats["frames_dropped_overflow"] += 1 - def get_nowait(self) -> Optional[object]: + def get_nowait(self) -> object | None: return self.get(timeout=0.0) - def get(self, timeout: Optional[float] = None) -> Optional[object]: + def get(self, timeout: float | None = None) -> object | None: if self._stop_after_current: self._stop_after_current = False return _STOP @@ -1357,7 +1358,7 @@ def _accept_candidate(self, candidate: object) -> bool: return media_advance_s >= self._time_debt_s @staticmethod - def _frame_media_time_s(frame: object) -> Optional[float]: + def _frame_media_time_s(frame: object) -> float | None: pts = getattr(frame, "pts", None) time_base = getattr(frame, "time_base", None) if pts is None or time_base is None: diff --git a/src/livepeer_gateway/remote_signer.py b/src/livepeer_gateway/remote_signer.py index 6196aea..81f6ab1 100644 --- a/src/livepeer_gateway/remote_signer.py +++ b/src/livepeer_gateway/remote_signer.py @@ -3,7 +3,7 @@ import asyncio import logging from dataclasses import dataclass, replace -from typing import Any, Optional +from typing import Any from .async_cache import async_lru_cache from .errors import ( @@ -21,7 +21,7 @@ @dataclass(frozen=True) class GetPaymentResponse: payment: str - seg_creds: Optional[str] = None + seg_creds: str | None = None @dataclass(frozen=True) @@ -48,15 +48,15 @@ class SignerMaterial: class RemoteSignerError(LivepeerGatewayError): signer_url: str message: str - cause: Optional[BaseException] = None + cause: BaseException | None = None def __str__(self) -> str: return f"Remote signer error: {self.message} (url={self.signer_url})" def _freeze_headers( - headers: Optional[dict[str, str]], -) -> Optional[frozenset[tuple[str, str]]]: + headers: dict[str, str] | None, +) -> frozenset[tuple[str, str]] | None: """Convert a headers dict to a hashable cache key.""" if headers is None: return None diff --git a/src/livepeer_gateway/segment_reader.py b/src/livepeer_gateway/segment_reader.py index 7250505..91c9ea0 100644 --- a/src/livepeer_gateway/segment_reader.py +++ b/src/livepeer_gateway/segment_reader.py @@ -2,7 +2,7 @@ import asyncio from dataclasses import dataclass -from typing import Optional, Protocol +from typing import Protocol import aiohttp @@ -51,12 +51,12 @@ class _SegmentBuffer: """ source: AsyncByteSource producer_read_size: int = 32 * 1024 # size of each source.read request - max_bytes: Optional[int] = None # safety bound on TOTAL buffered bytes + max_bytes: int | None = None # safety bound on TOTAL buffered bytes def __post_init__(self) -> None: self._buf = bytearray() self._eof = False - self._error: Optional[BaseException] = None + self._error: BaseException | None = None self._stats: dict[str, int] = { "chunks_read": 0, "bytes_read": 0, @@ -111,7 +111,7 @@ async def _produce_chunk(self) -> None: self._cond.notify_all() - async def _read_for(self, cursor: int, n: int) -> Optional[bytes]: + async def _read_for(self, cursor: int, n: int) -> bytes | None: """ Return up to n bytes starting at cursor. Returns None on EOF/closed. @@ -162,7 +162,7 @@ def __init__(self, writer: _SegmentBuffer): self._w = writer self._pos = 0 - async def read(self, chunk_size: int = 32 * 1024) -> Optional[bytes]: + async def read(self, chunk_size: int = 32 * 1024) -> bytes | None: out = await self._w._read_for(self._pos, chunk_size) if out is None: return None @@ -181,7 +181,7 @@ def __init__( self, response: aiohttp.ClientResponse, *, - max_bytes: Optional[int] = 10 * 1024 * 1024, + max_bytes: int | None = 10 * 1024 * 1024, producer_read_size: int = 32 * 1024, ): self.response = response @@ -208,7 +208,7 @@ def eos(self) -> bool: # Header-based end-of-segment marker (your existing behavior). return self.response.headers.get("Lp-Trickle-Closed") is not None - def headers(self) -> "aiohttp.typedefs.LooseHeaders": + def headers(self) -> aiohttp.typedefs.LooseHeaders: return self.response.headers async def close(self) -> None: diff --git a/src/livepeer_gateway/token.py b/src/livepeer_gateway/token.py index 11f354c..ce7af29 100644 --- a/src/livepeer_gateway/token.py +++ b/src/livepeer_gateway/token.py @@ -3,7 +3,7 @@ import base64 import binascii import json -from typing import Any, Optional +from typing import Any from .errors import LivepeerGatewayError @@ -43,7 +43,7 @@ def parse_token(token: str) -> dict[str, Any]: if orchestrators is not None and not isinstance(orchestrators, list): raise LivepeerGatewayError("Invalid token: orchestrators must be an array of strings") - normalized_orchestrators: Optional[list[str]] = None + normalized_orchestrators: list[str] | None = None if isinstance(orchestrators, list): normalized_orchestrators = [] for item in orchestrators: diff --git a/src/livepeer_gateway/trickle_publisher.py b/src/livepeer_gateway/trickle_publisher.py index 7b64ee7..2ecd4c2 100644 --- a/src/livepeer_gateway/trickle_publisher.py +++ b/src/livepeer_gateway/trickle_publisher.py @@ -4,7 +4,7 @@ from dataclasses import dataclass import logging import time -from typing import Optional, AsyncIterator, Callable +from collections.abc import AsyncIterator, Callable import aiohttp @@ -63,8 +63,8 @@ def __init__( message: str, *, seq: int, - url: Optional[str] = None, - status: Optional[int] = None, + url: str | None = None, + status: int | None = None, ) -> None: super().__init__(message) self.seq = seq @@ -80,7 +80,7 @@ def __init__( message: str, *, consecutive_failures: int, - url: Optional[str] = None, + url: str | None = None, ) -> None: super().__init__(message) self.consecutive_failures = consecutive_failures @@ -116,19 +116,19 @@ def __init__( self.seq = int(start_seq) # Lazily initialized async runtime bits (safe to construct in sync code). - self._lock: Optional[asyncio.Lock] = None - self._session: Optional[aiohttp.ClientSession] = None + self._lock: asyncio.Lock | None = None + self._session: aiohttp.ClientSession | None = None self._closing = False self._closed = False # Preconnected writer state for the next segment. - self._next_state: Optional[_SegmentPostState] = None - self._preconnect_task_handle: Optional[asyncio.Task[None]] = None + self._next_state: _SegmentPostState | None = None + self._preconnect_task_handle: asyncio.Task[None] | None = None self._post_tasks: set[asyncio.Task[None]] = set() # Terminal failure for the whole publisher. Once set, no new segments # should be opened or written. - self._terminal_error: Optional[TricklePublisherTerminalError] = None + self._terminal_error: TricklePublisherTerminalError | None = None self._consecutive_failures: int = 0 self._started_at = time.time() self._stats: dict[str, int] = { @@ -147,7 +147,7 @@ def __init__( "empty_segments_completed": 0, } - async def __aenter__(self) -> "TricklePublisher": + async def __aenter__(self) -> TricklePublisher: return self async def __aexit__(self, exc_type, exc_value, traceback) -> None: @@ -201,9 +201,9 @@ async def _run_post(self, url: str, seg_state: _SegmentPostState) -> None: assert self._session is not None seq = seg_state.seq - final_exc: Optional[TrickleSegmentWriteError] = None - final_status: Optional[int] = None - final_body: Optional[str] = None + final_exc: TrickleSegmentWriteError | None = None + final_status: int | None = None + final_body: str | None = None for attempt in range(2): self._stats["post_attempts"] += 1 @@ -359,7 +359,7 @@ async def _resolve_next_seq(self) -> int: _LOG.warning("Trickle /next request failed", exc_info=True) return -1 - async def next(self) -> "SegmentWriter": + async def next(self) -> SegmentWriter: # Fail fast via the publisher error hierarchy, not a generic RuntimeError. if self._closing or self._closed: raise TricklePublisherTerminalError( @@ -545,9 +545,9 @@ class _SegmentPostState: def __init__(self, seq: int, *, send_reset: bool = False) -> None: self.seq = seq - self.queue: asyncio.Queue[Optional[bytes]] = asyncio.Queue(maxsize=1) + self.queue: asyncio.Queue[bytes | None] = asyncio.Queue(maxsize=1) # Failure for this one segment only, not necessarily terminal - self.error: Optional[TrickleSegmentWriteError] = None + self.error: TrickleSegmentWriteError | None = None self.data_consumed: bool = False self.send_reset: bool = send_reset @@ -558,9 +558,9 @@ def __init__( self, seg_state: _SegmentPostState, *, - error_getter: Optional[Callable[[], Optional[TricklePublisherTerminalError]]] = None, - on_write_bytes: Optional[Callable[[int], None]] = None, - on_write_timeout: Optional[Callable[[], None]] = None, + error_getter: Callable[[], TricklePublisherTerminalError | None] | None = None, + on_write_bytes: Callable[[int], None] | None = None, + on_write_timeout: Callable[[], None] | None = None, ): self._seg_state = seg_state self.queue = seg_state.queue @@ -583,7 +583,7 @@ async def write(self, data: bytes) -> None: await asyncio.wait_for(self.queue.put(data), timeout=_SEGMENT_QUEUE_PUT_TIMEOUT_S) if self._on_write_bytes is not None: self._on_write_bytes(len(data)) - except asyncio.TimeoutError as e: + except TimeoutError as e: if self._on_write_timeout is not None: self._on_write_timeout() if self._error_getter is not None: @@ -605,7 +605,7 @@ async def close(self) -> None: except BaseException: _LOG.warning("Trickle segment close suppressed seq=%s", self._seq, exc_info=True) - async def __aenter__(self) -> "SegmentWriter": + async def __aenter__(self) -> SegmentWriter: return self async def __aexit__(self, exc_type, exc_value, traceback) -> None: diff --git a/src/livepeer_gateway/trickle_subscriber.py b/src/livepeer_gateway/trickle_subscriber.py index a8add94..f56f7f4 100644 --- a/src/livepeer_gateway/trickle_subscriber.py +++ b/src/livepeer_gateway/trickle_subscriber.py @@ -4,7 +4,6 @@ from dataclasses import dataclass import logging import time -from typing import Optional import aiohttp @@ -61,7 +60,7 @@ def __init__( start_seq: int = -2, max_retries: int = 5, connection_close: bool = False, - max_bytes: Optional[int] = None, + max_bytes: int | None = None, ): if max_bytes is not None and max_bytes <= 0: raise ValueError("max_bytes must be > 0") @@ -71,13 +70,13 @@ def __init__( self._connection_close = connection_close self._max_bytes = max_bytes - self._pending_get: Optional[aiohttp.ClientResponse] = None - self._lock: Optional[asyncio.Lock] = None - self._session: Optional[aiohttp.ClientSession] = None + self._pending_get: aiohttp.ClientResponse | None = None + self._lock: asyncio.Lock | None = None + self._session: aiohttp.ClientSession | None = None self._errored = False self._closing = False self._closed = False - self._prefetch_task: Optional[asyncio.Task[None]] = None + self._prefetch_task: asyncio.Task[None] | None = None self._started_at = time.time() self._stats: dict[str, int] = { "get_attempts": 0, @@ -91,7 +90,7 @@ def __init__( "latest_seq": start_seq, } - async def __aenter__(self) -> "TrickleSubscriber": + async def __aenter__(self) -> TrickleSubscriber: return self async def __aexit__(self, exc_type, exc_value, traceback) -> None: @@ -110,7 +109,7 @@ def _segment_url(self, seq: int) -> str: return f"{self.base_url}/{seq}" @staticmethod - def _latest_seq(headers: "aiohttp.typedefs.LooseHeaders", current_seq: int) -> int: + def _latest_seq(headers: aiohttp.typedefs.LooseHeaders, current_seq: int) -> int: latest_header = headers.get("Lp-Trickle-Latest") if latest_header is None: return current_seq @@ -119,7 +118,7 @@ def _latest_seq(headers: "aiohttp.typedefs.LooseHeaders", current_seq: int) -> i except ValueError: return current_seq - async def _preconnect(self) -> Optional[aiohttp.ClientResponse]: + async def _preconnect(self) -> aiohttp.ClientResponse | None: """ Preconnect to the server by making a GET request to fetch the next segment. @@ -203,7 +202,7 @@ async def _preconnect(self) -> Optional[aiohttp.ClientResponse]: self._errored = True return None - async def next(self) -> Optional["SegmentReader"]: + async def next(self) -> SegmentReader | None: """Retrieve data from the current segment and set up the next segment concurrently.""" if self._closing or self._closed: return None