From 561ac7bee3185752380435f8961e9b4f5b0697aa Mon Sep 17 00:00:00 2001 From: snus-kin Date: Mon, 24 Aug 2026 12:35:27 +0100 Subject: [PATCH 1/3] chore: update prek deps --- .pre-commit-config.yaml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index f6732d9..d5e20ec 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -1,12 +1,12 @@ repos: - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.12.7 + rev: v0.16.4 hooks: - id: ruff-check args: [--fix] - id: ruff-format - repo: https://github.com/astral-sh/uv-pre-commit - rev: 0.7.15 + rev: 0.12.5 hooks: - id: uv-lock - repo: local @@ -18,7 +18,7 @@ repos: types_or: [python, pyi] pass_filenames: false - repo: https://github.com/pre-commit/pre-commit-hooks - rev: v5.0.0 + rev: v6.0.0 hooks: - id: mixed-line-ending - id: end-of-file-fixer From dede9ad3c1d6b5b1993817ae6da103a4b513ad34 Mon Sep 17 00:00:00 2001 From: snus-kin Date: Mon, 24 Aug 2026 12:44:42 +0100 Subject: [PATCH 2/3] chore: ignore CPY001 copyright-header rule in ruff config Project files don't carry copyright headers; ignore the rule instead of adding headers repo-wide. --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 6a9cb63..e37d362 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,7 +65,7 @@ exclude = ["src/resolver_athena_client/generated/*", "docs"] [tool.ruff.lint] select = ["ALL"] -ignore = ["COM812", "D213", "D211", "D203", "S324", "ASYNC109"] +ignore = ["COM812", "D213", "D211", "D203", "S324", "ASYNC109", "CPY001"] [tool.ruff.lint.per-file-ignores] # Ignore doc lint rules in tests. From 6c8e93f3034276b9d97c76ad4dc09236c9f9ff56 Mon Sep 17 00:00:00 2001 From: snus-kin Date: Mon, 24 Aug 2026 12:46:09 +0100 Subject: [PATCH 3/3] fix: resolve PLR0913/PLR0911/PLR2004/TRY301 lint findings without noqa Replaces the WorkerBatcher noqa: PLR0913 suppression by grouping its tuning knobs into a WorkerBatcherOptions dataclass, rewrites detect_image_format as a table of magic-number checks to drop the noqa: PLR0911 return-count suppression, names the magic constant in test_image_data instead of noqa: PLR2004, and hoists the CancelledError raise out of the try block in test_timeout_behavior to satisfy TRY301 without a noqa. Left the PERF203/BLE001 noqas in place where they mark deliberate control-flow and callback-contract exceptions. --- .../client/athena_client.py | 9 +- .../client/image_format_detector.py | 73 ++++++-------- .../client/transformers/worker_batcher.py | 41 ++++---- tests/client/models/test_image_data.py | 6 +- tests/client/test_athena_client.py | 10 +- tests/client/test_timeout_behavior.py | 8 +- .../test_worker_batcher_simple.py | 97 ++++++++++++------- 7 files changed, 137 insertions(+), 107 deletions(-) diff --git a/src/resolver_athena_client/client/athena_client.py b/src/resolver_athena_client/client/athena_client.py index 80cc504..a137731 100644 --- a/src/resolver_athena_client/client/athena_client.py +++ b/src/resolver_athena_client/client/athena_client.py @@ -18,6 +18,7 @@ ) from resolver_athena_client.client.transformers.worker_batcher import ( WorkerBatcher, + WorkerBatcherOptions, ) from resolver_athena_client.generated.athena.models_pb2 import ( ClassificationInput, @@ -289,9 +290,11 @@ async def transform_image(image_data: ImageData) -> ClassificationInput: source=images, transformer_func=transform_image, deployment_id=self.options.deployment_id, - max_batch_size=self.options.max_batch_size, - num_workers=self.options.num_workers, - keepalive_interval=self.options.keepalive_interval or 30.0, + options=WorkerBatcherOptions( + max_batch_size=self.options.max_batch_size, + num_workers=self.options.num_workers, + keepalive_interval=self.options.keepalive_interval or 30.0, + ), ) # Track the worker for cleanup diff --git a/src/resolver_athena_client/client/image_format_detector.py b/src/resolver_athena_client/client/image_format_detector.py index ca836d9..195b615 100644 --- a/src/resolver_athena_client/client/image_format_detector.py +++ b/src/resolver_athena_client/client/image_format_detector.py @@ -1,5 +1,7 @@ """Utility for detecting image formats from raw bytes.""" +from collections.abc import Callable + from resolver_athena_client.generated.athena.models_pb2 import ImageFormat PNG_MAGIC_BYTES = b"\x89PNG" @@ -13,7 +15,32 @@ TIFF_BE_MAGIC_BYTES = b"MM\x00*" -def detect_image_format(data: bytes) -> ImageFormat.ValueType: # noqa: PLR0911 +def _is_webp(data: bytes) -> bool: + """Check for the RIFF....WEBP signature (12 bytes minimum).""" + return ( + data[:4] == WEBP_RIFF_MAGIC_BYTES + and data[8:12] == WEBP_WEBP_MAGIC_BYTES + ) + + +_ImageFormatDetector = tuple[Callable[[bytes], bool], ImageFormat.ValueType] +_FORMAT_DETECTORS: list[_ImageFormatDetector] = [ + (lambda d: d.startswith(PNG_MAGIC_BYTES), ImageFormat.IMAGE_FORMAT_PNG), + (lambda d: d.startswith(JPEG_MAGIC_BYTES), ImageFormat.IMAGE_FORMAT_JPEG), + ( + lambda d: d.startswith((GIF87A_MAGIC_BYTES, GIF89A_MAGIC_BYTES)), + ImageFormat.IMAGE_FORMAT_GIF, + ), + (lambda d: d.startswith(BMP_MAGIC_BYTES), ImageFormat.IMAGE_FORMAT_BMP), + (_is_webp, ImageFormat.IMAGE_FORMAT_WEBP), + ( + lambda d: d.startswith((TIFF_LE_MAGIC_BYTES, TIFF_BE_MAGIC_BYTES)), + ImageFormat.IMAGE_FORMAT_TIFF, + ), +] + + +def detect_image_format(data: bytes) -> ImageFormat.ValueType: """Detect image format from raw bytes using magic number signatures. Args: @@ -28,46 +55,8 @@ def detect_image_format(data: bytes) -> ImageFormat.ValueType: # noqa: PLR0911 if not data: return ImageFormat.IMAGE_FORMAT_UNSPECIFIED - # Check magic numbers for common image formats - # PNG: starts with PNG_MAGIC_BYTES - png_len = len(PNG_MAGIC_BYTES) - if len(data) >= png_len and data[:png_len] == PNG_MAGIC_BYTES: - return ImageFormat.IMAGE_FORMAT_PNG - - # JPEG: starts with JPEG_MAGIC_BYTES - jpeg_len = len(JPEG_MAGIC_BYTES) - if len(data) >= jpeg_len and data[:jpeg_len] == JPEG_MAGIC_BYTES: - return ImageFormat.IMAGE_FORMAT_JPEG - - # GIF: starts with GIF87A_MAGIC_BYTES or GIF89A_MAGIC_BYTES - gif_len = len(GIF87A_MAGIC_BYTES) - if len(data) >= gif_len and data[:gif_len] in ( - GIF87A_MAGIC_BYTES, - GIF89A_MAGIC_BYTES, - ): - return ImageFormat.IMAGE_FORMAT_GIF - - # BMP: starts with BMP_MAGIC_BYTES - bmp_len = len(BMP_MAGIC_BYTES) - if len(data) >= bmp_len and data[:bmp_len] == BMP_MAGIC_BYTES: - return ImageFormat.IMAGE_FORMAT_BMP - - # WebP: RIFF....WEBP (12 bytes minimum for full signature) - webp_min_len = len(WEBP_RIFF_MAGIC_BYTES) + len(WEBP_WEBP_MAGIC_BYTES) + 4 - if ( - len(data) >= webp_min_len - and data[:4] == WEBP_RIFF_MAGIC_BYTES - and data[8:12] == WEBP_WEBP_MAGIC_BYTES - ): - return ImageFormat.IMAGE_FORMAT_WEBP - - # TIFF: little-endian or big-endian magic bytes - tiff_len = len(TIFF_LE_MAGIC_BYTES) - if len(data) >= tiff_len and ( - data[:tiff_len] == TIFF_LE_MAGIC_BYTES - or data[:tiff_len] == TIFF_BE_MAGIC_BYTES - ): - return ImageFormat.IMAGE_FORMAT_TIFF + for matches, image_format in _FORMAT_DETECTORS: + if matches(data): + return image_format - # Fallback when format cannot be determined return ImageFormat.IMAGE_FORMAT_UNSPECIFIED diff --git a/src/resolver_athena_client/client/transformers/worker_batcher.py b/src/resolver_athena_client/client/transformers/worker_batcher.py index 6218aa0..75fcd5f 100644 --- a/src/resolver_athena_client/client/transformers/worker_batcher.py +++ b/src/resolver_athena_client/client/transformers/worker_batcher.py @@ -5,6 +5,7 @@ import logging import time from collections.abc import AsyncIterator, Awaitable, Callable +from dataclasses import dataclass from typing import Generic, TypeVar from resolver_athena_client.generated.athena.models_pb2 import ( @@ -15,19 +16,26 @@ T = TypeVar("T") +@dataclass +class WorkerBatcherOptions: + """Tuning options for `WorkerBatcher`.""" + + max_batch_size: int = 10 + num_workers: int = 4 + queue_size: int = 100 + keepalive_interval: float = 30.0 + batch_timeout: float = 0.1 + + class WorkerBatcher(Generic[T]): """Asyncio worker-based batcher with concurrent processing and buffering.""" - def __init__( # noqa: PLR0913 + def __init__( self, source: AsyncIterator[T], transformer_func: Callable[[T], Awaitable[ClassificationInput]], deployment_id: str, - max_batch_size: int = 10, - num_workers: int = 4, - queue_size: int = 100, - keepalive_interval: float = 30.0, - batch_timeout: float = 0.1, + options: WorkerBatcherOptions | None = None, ) -> None: """Initialize the worker batcher. @@ -37,29 +45,28 @@ def __init__( # noqa: PLR0913 transformer_func: Function to transform items (e.g., image processing) deployment_id: Deployment ID for requests - max_batch_size: Maximum items per batch - num_workers: Number of concurrent worker tasks - queue_size: Size of internal processing queue - keepalive_interval: Seconds between keepalive requests - batch_timeout: Max seconds to wait before sending partial batch + options: Tuning options for batch size, worker count, and + timing. Defaults to `WorkerBatcherOptions()`. """ + options = options or WorkerBatcherOptions() + self.source: AsyncIterator[T] = source self.transformer_func: Callable[[T], Awaitable[ClassificationInput]] = ( transformer_func ) self.deployment_id: str = deployment_id - self.max_batch_size: int = max_batch_size - self.num_workers: int = num_workers - self.keepalive_interval: float = keepalive_interval - self.batch_timeout: float = batch_timeout + self.max_batch_size: int = options.max_batch_size + self.num_workers: int = options.num_workers + self.keepalive_interval: float = options.keepalive_interval + self.batch_timeout: float = options.batch_timeout # Internal queues and state - use Optional[T] to handle None sentinel self.input_queue: asyncio.Queue[T | None] = asyncio.Queue( - maxsize=queue_size + maxsize=options.queue_size ) self.output_queue: asyncio.Queue[ClassificationInput] = asyncio.Queue( - maxsize=queue_size + maxsize=options.queue_size ) self.processed_items: list[ClassificationInput] = [] diff --git a/tests/client/models/test_image_data.py b/tests/client/models/test_image_data.py index 16d97f3..a9299dd 100644 --- a/tests/client/models/test_image_data.py +++ b/tests/client/models/test_image_data.py @@ -5,6 +5,8 @@ from resolver_athena_client.client.models import ImageData from resolver_athena_client.generated.athena.models_pb2 import ImageFormat +EXPECTED_HASH_COUNT = 2 + def test_image_data_detects_png_format() -> None: """Test that PNG format is detected during initialization.""" @@ -78,8 +80,8 @@ def test_image_data_transformation_preserves_format() -> None: # Format should still be PNG (transformers will update it if needed) assert image_data.image_format == ImageFormat.IMAGE_FORMAT_PNG - assert len(image_data.sha256_hashes) == 2 # noqa: PLR2004 - assert len(image_data.md5_hashes) == 2 # noqa: PLR2004 + assert len(image_data.sha256_hashes) == EXPECTED_HASH_COUNT + assert len(image_data.md5_hashes) == EXPECTED_HASH_COUNT @pytest.mark.parametrize( diff --git a/tests/client/test_athena_client.py b/tests/client/test_athena_client.py index f5aaae9..519a4b7 100644 --- a/tests/client/test_athena_client.py +++ b/tests/client/test_athena_client.py @@ -2,7 +2,7 @@ import asyncio import contextlib -from typing import cast +from typing import TYPE_CHECKING, cast from unittest import mock import pytest @@ -23,6 +23,11 @@ ) from tests.utils.mock_async_iterator import MockAsyncIterator +if TYPE_CHECKING: + from resolver_athena_client.client.transformers.worker_batcher import ( + WorkerBatcherOptions, + ) + @pytest.fixture def mock_channel() -> mock.Mock: @@ -388,7 +393,8 @@ async def start_classification() -> None: # Verify WorkerBatcher was created with correct num_workers mock_worker_batcher_cls.assert_called_once() call_kwargs = mock_worker_batcher_cls.call_args.kwargs - assert call_kwargs["num_workers"] == custom_num_workers + batcher_options = cast("WorkerBatcherOptions", call_kwargs["options"]) + assert batcher_options.num_workers == custom_num_workers @pytest.mark.asyncio diff --git a/tests/client/test_timeout_behavior.py b/tests/client/test_timeout_behavior.py index 8feeddd..5abebc0 100644 --- a/tests/client/test_timeout_behavior.py +++ b/tests/client/test_timeout_behavior.py @@ -325,11 +325,11 @@ async def test_timeout_with_cancellation() -> None: responses: list[ClassifyResponse] = [] classify_task = None - try: + def cancel_after_target() -> None: + """Cancel processing after target responses.""" + raise asyncio.CancelledError - def cancel_after_target() -> None: - """Cancel processing after target responses.""" - raise asyncio.CancelledError # noqa: TRY301 + try: async def collect_responses() -> None: response_iter = aiter(client.classify_images(image_stream)) diff --git a/tests/client/transformers/test_worker_batcher_simple.py b/tests/client/transformers/test_worker_batcher_simple.py index a022e4d..15976ce 100644 --- a/tests/client/transformers/test_worker_batcher_simple.py +++ b/tests/client/transformers/test_worker_batcher_simple.py @@ -6,6 +6,7 @@ from resolver_athena_client.client.transformers.worker_batcher import ( WorkerBatcher, + WorkerBatcherOptions, ) from resolver_athena_client.generated.athena.models_pb2 import ( ClassificationInput, @@ -104,9 +105,12 @@ async def test_worker_batcher_basic() -> None: source=source, transformer_func=identity_transform, deployment_id="test-deployment", - batch_timeout=0.001, - keepalive_interval=0.001, # Very short keepalive for immediate response - num_workers=1, # Single worker for simple case + # Very short keepalive/timeout for immediate response, single worker + options=WorkerBatcherOptions( + batch_timeout=0.001, + keepalive_interval=0.001, + num_workers=1, + ), ) # Should get one request with the item @@ -135,10 +139,12 @@ async def test_worker_batcher_batching() -> None: source=source, transformer_func=identity_transform, deployment_id="test-deployment", - max_batch_size=BATCH_SIZE_TWO, - batch_timeout=0.001, - keepalive_interval=0.1, - num_workers=1, + options=WorkerBatcherOptions( + max_batch_size=BATCH_SIZE_TWO, + batch_timeout=0.001, + keepalive_interval=0.1, + num_workers=1, + ), ) # Should get first batch with 2 items @@ -169,10 +175,13 @@ async def test_worker_batcher_timeout( source=source_with_timeout, transformer_func=identity_transform, deployment_id="test-deployment", - max_batch_size=BATCH_SIZE_THREE, - batch_timeout=0.001, # Very short timeout to trigger timeout behavior - keepalive_interval=0.1, - num_workers=1, + # Very short timeout to trigger timeout behavior + options=WorkerBatcherOptions( + max_batch_size=BATCH_SIZE_THREE, + batch_timeout=0.001, + keepalive_interval=0.1, + num_workers=1, + ), ) # Should get partial batches due to timeout @@ -213,9 +222,11 @@ async def test_worker_batcher_empty() -> None: source=source, transformer_func=identity_transform, deployment_id="test-deployment", - batch_timeout=0.001, - keepalive_interval=0.001, - num_workers=1, + options=WorkerBatcherOptions( + batch_timeout=0.001, + keepalive_interval=0.001, + num_workers=1, + ), ) # Empty source should produce keepalive @@ -238,10 +249,12 @@ async def test_worker_batcher_exact_batch() -> None: source=source, transformer_func=identity_transform, deployment_id="test-deployment", - max_batch_size=BATCH_SIZE_THREE, - batch_timeout=0.001, - keepalive_interval=0.001, - num_workers=1, + options=WorkerBatcherOptions( + max_batch_size=BATCH_SIZE_THREE, + batch_timeout=0.001, + keepalive_interval=0.001, + num_workers=1, + ), ) # Should get exactly one batch @@ -279,10 +292,12 @@ async def test_worker_batcher_edge_cases() -> None: source=source, transformer_func=identity_transform, deployment_id="test-deployment", - max_batch_size=BATCH_SIZE_TWO, - batch_timeout=0.001, - keepalive_interval=0.1, - num_workers=1, + options=WorkerBatcherOptions( + max_batch_size=BATCH_SIZE_TWO, + batch_timeout=0.001, + keepalive_interval=0.1, + num_workers=1, + ), ) total_items_received = 0 @@ -327,10 +342,12 @@ async def test_worker_batcher_full_batch() -> None: source=source, transformer_func=identity_transform, deployment_id="test-deployment", - max_batch_size=FULL_BATCH_SIZE, - batch_timeout=0.001, - keepalive_interval=0.1, - num_workers=1, + options=WorkerBatcherOptions( + max_batch_size=FULL_BATCH_SIZE, + batch_timeout=0.001, + keepalive_interval=0.1, + num_workers=1, + ), ) # Should get first full batch @@ -364,10 +381,12 @@ async def test_worker_batcher_source_iteration_end() -> None: source=source, transformer_func=identity_transform, deployment_id="test-deployment", - max_batch_size=BATCH_SIZE_THREE, - batch_timeout=0.001, - keepalive_interval=0.001, - num_workers=1, + options=WorkerBatcherOptions( + max_batch_size=BATCH_SIZE_THREE, + batch_timeout=0.001, + keepalive_interval=0.001, + num_workers=1, + ), ) # Should get a batch with available items (less than max_batch_size) @@ -401,10 +420,12 @@ async def test_worker_batcher_iterator_end_no_timeout() -> None: source=source, transformer_func=identity_transform, deployment_id="test-deployment", - max_batch_size=BATCH_SIZE_TWO, - batch_timeout=fast_timeout, - keepalive_interval=0.1, - num_workers=1, + options=WorkerBatcherOptions( + max_batch_size=BATCH_SIZE_TWO, + batch_timeout=fast_timeout, + keepalive_interval=0.1, + num_workers=1, + ), ) # Should get the item before timeout @@ -437,9 +458,11 @@ async def modify_transform( source=source, transformer_func=modify_transform, deployment_id="test-deployment", - batch_timeout=0.001, - keepalive_interval=0.1, - num_workers=1, + options=WorkerBatcherOptions( + batch_timeout=0.001, + keepalive_interval=0.1, + num_workers=1, + ), ) # Should get transformed item