Fix all ruff and mypy issues; CI passes clean

Author: janbjorge

jqueue/__init__.py

@@ -63,7 +63,12 @@ async def write(content, if_match=None) -> str
 from jqueue.core.broker import BrokerQueue
 from jqueue.core.direct import DirectQueue
 from jqueue.core.heartbeat import HeartbeatManager
-from jqueue.domain.errors import CASConflictError, JQueueError, JobNotFoundError, StorageError
+from jqueue.domain.errors import (
+    CASConflictError,
+    JobNotFoundError,
+    JQueueError,
+    StorageError,
+)
 from jqueue.domain.models import Job, JobStatus, QueueState
 from jqueue.ports.storage import ObjectStoragePort
 

jqueue/adapters/storage/gcs.py

@@ -18,6 +18,7 @@
 Note: google-cloud-storage is synchronous. All operations are wrapped in
 asyncio.to_thread to avoid blocking the event loop.
 """
+
 from __future__ import annotations
 
 import asyncio
@@ -50,13 +51,13 @@ def _get_client(self) -> GCSClient:
         if self.client is not None:
             return self.client
         try:
-            from google.cloud import storage  # type: ignore[import-untyped]
+            from google.cloud import storage
         except ImportError as exc:
             raise ImportError(
                 "GCSStorage requires google-cloud-storage. "
                 "Install with: pip install 'jqueue[gcs]'"
             ) from exc
-        return storage.Client()  # type: ignore[return-value]
+        return storage.Client()
 
     async def read(self) -> tuple[bytes, str | None]:
         """Read the state blob. Returns (b"", None) if the blob does not exist."""
@@ -86,15 +87,15 @@ async def write(
 
     def _sync_read(self) -> tuple[bytes, str | None]:
         try:
-            from google.api_core import exceptions as gapi_exc  # type: ignore[import-untyped]
+            from google.api_core import exceptions as gapi_exc
         except ImportError as exc:
             raise ImportError(
                 "GCSStorage requires google-cloud-storage. "
                 "Install with: pip install 'jqueue[gcs]'"
             ) from exc
 
         client = self._get_client()
-        blob = client.bucket(self.bucket_name).blob(self.blob_name)  # type: ignore[attr-defined]
+        blob = client.bucket(self.bucket_name).blob(self.blob_name)
         try:
             content: bytes = blob.download_as_bytes()
             return content, str(blob.generation)
@@ -103,27 +104,27 @@ def _sync_read(self) -> tuple[bytes, str | None]:
 
     def _sync_write(self, content: bytes, if_match: str | None) -> str:
         try:
-            from google.api_core import exceptions as gapi_exc  # type: ignore[import-untyped]
+            from google.api_core import exceptions as gapi_exc
         except ImportError as exc:
             raise ImportError(
                 "GCSStorage requires google-cloud-storage. "
                 "Install with: pip install 'jqueue[gcs]'"
             ) from exc
 
         client = self._get_client()
-        blob = client.bucket(self.bucket_name).blob(self.blob_name)  # type: ignore[attr-defined]
+        blob = client.bucket(self.bucket_name).blob(self.blob_name)
 
         # if_generation_match=0 → "blob must not exist yet"
         gen_match: int = 0 if if_match is None else int(if_match)
 
         try:
-            blob.upload_from_string(  # type: ignore[attr-defined]
+            blob.upload_from_string(
                 content,
                 content_type="application/json",
                 if_generation_match=gen_match,
             )
         except gapi_exc.PreconditionFailed as exc:
             raise CASConflictError("GCS generation mismatch") from exc
 
-        blob.reload()  # type: ignore[attr-defined]
+        blob.reload()
         return str(blob.generation)

jqueue/adapters/storage/s3.py

@@ -17,6 +17,7 @@
 Compatible with S3-compatible storage that supports conditional writes:
   MinIO, Cloudflare R2, Tigris, etc.
 """
+
 from __future__ import annotations
 
 import dataclasses
@@ -52,12 +53,12 @@ def _get_session(self) -> AioBoto3Session:
         if self.session is not None:
             return self.session
         try:
-            import aioboto3  # type: ignore[import-untyped]
+            import aioboto3
         except ImportError as exc:
             raise ImportError(
                 "S3Storage requires aioboto3. Install with: pip install 'jqueue[s3]'"
             ) from exc
-        return aioboto3.Session()  # type: ignore[return-value]
+        return aioboto3.Session()
 
     def _client_kwargs(self) -> dict[str, str]:
         """Build kwargs forwarded to the S3 client constructor."""
@@ -72,7 +73,7 @@ async def read(self) -> tuple[bytes, str | None]:
         """Read the state object. Returns (b"", None) if the key does not exist."""
         session = self._get_session()
         try:
-            async with session.client("s3", **self._client_kwargs()) as s3:  # type: ignore[attr-defined]
+            async with session.client("s3", **self._client_kwargs()) as s3:
                 try:
                     response = await s3.get_object(Bucket=self.bucket, Key=self.key)
                     content: bytes = await response["Body"].read()
@@ -95,7 +96,7 @@ async def write(
         """CAS write. Raises CASConflictError on ETag mismatch (PreconditionFailed)."""
         session = self._get_session()
         try:
-            async with session.client("s3", **self._client_kwargs()) as s3:  # type: ignore[attr-defined]
+            async with session.client("s3", **self._client_kwargs()) as s3:
                 put_kwargs: dict[str, str | bytes] = {
                     "Bucket": self.bucket,
                     "Key": self.key,

jqueue/core/broker.py

@@ -57,7 +57,7 @@ def __post_init__(self) -> None:
             stale_timeout=self.stale_timeout,
         )
 
-    async def __aenter__(self) -> "BrokerQueue":
+    async def __aenter__(self) -> BrokerQueue:
         await self._loop.start()
         return self
 

jqueue/core/direct.py

@@ -19,8 +19,8 @@
 
 import asyncio
 import dataclasses
-from datetime import datetime, timedelta, timezone
-from typing import Callable
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
 
 from jqueue.core import codec
 from jqueue.domain.errors import CASConflictError, JobNotFoundError
@@ -78,7 +78,7 @@ def _fn(state: QueueState) -> QueueState:
             new_state = state
             for job in available:
                 updated = job.with_status(JobStatus.IN_PROGRESS).with_heartbeat(
-                    datetime.now(timezone.utc)
+                    datetime.now(UTC)
                 )
                 new_state = new_state.with_job_replaced(updated)
                 claimed.append(updated)
@@ -112,7 +112,7 @@ def _fn(state: QueueState) -> QueueState:
             if job is None:
                 raise JobNotFoundError(job_id)
             return state.with_job_replaced(
-                job.with_heartbeat(datetime.now(timezone.utc))
+                job.with_heartbeat(datetime.now(UTC))
             )
 
         await self._mutate(_fn)
@@ -123,7 +123,7 @@ async def requeue_stale(self, timeout: timedelta) -> int:
 
         Returns the number of jobs re-queued.
         """
-        cutoff = datetime.now(timezone.utc) - timeout
+        cutoff = datetime.now(UTC) - timeout
         requeued = 0
 
         def _fn(state: QueueState) -> QueueState:

jqueue/core/group_commit.py

@@ -27,8 +27,8 @@
 
 import asyncio
 import dataclasses
-from datetime import datetime, timedelta, timezone
-from typing import Callable
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
 
 from jqueue.core import codec
 from jqueue.domain.errors import CASConflictError, JQueueError
@@ -133,7 +133,7 @@ def _fn(state: QueueState) -> QueueState:
             new_state = state
             for job in available:
                 updated = job.with_status(JobStatus.IN_PROGRESS).with_heartbeat(
-                    datetime.now(timezone.utc)
+                    datetime.now(UTC)
                 )
                 new_state = new_state.with_job_replaced(updated)
                 claimed.append(updated)
@@ -169,7 +169,7 @@ def _fn(state: QueueState) -> QueueState:
             if job is None:
                 raise JobNotFoundError(job_id)
             return state.with_job_replaced(
-                job.with_heartbeat(datetime.now(timezone.utc))
+                job.with_heartbeat(datetime.now(UTC))
             )
 
         await self._submit(_fn)
@@ -224,7 +224,7 @@ async def _commit_batch(self, batch: list[_PendingOp]) -> None:
                 state = codec.decode(content)
 
                 # Sweep stale jobs on every write cycle (free — no extra I/O)
-                cutoff = datetime.now(timezone.utc) - self.stale_timeout
+                cutoff = datetime.now(UTC) - self.stale_timeout
                 state = state.requeue_stale(cutoff)
 
                 per_op_errors: dict[int, Exception] = {}

jqueue/core/heartbeat.py

@@ -59,7 +59,7 @@ class HeartbeatManager:
         default=None, init=False, repr=False
     )
 
-    async def __aenter__(self) -> "HeartbeatManager":
+    async def __aenter__(self) -> HeartbeatManager:
         self._task = asyncio.create_task(
             self._beat(), name=f"jqueue-heartbeat-{self.job_id}"
         )

jqueue/domain/models.py

@@ -13,13 +13,13 @@
 
 import base64
 import uuid
-from datetime import datetime, timezone
-from enum import Enum
+from datetime import UTC, datetime
+from enum import StrEnum
 
 from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator
 
 
-class JobStatus(str, Enum):
+class JobStatus(StrEnum):
     """Lifecycle states for a queued job."""
 
     QUEUED = "queued"
@@ -47,7 +47,7 @@ class Job(BaseModel):
     payload: bytes
     status: JobStatus = JobStatus.QUEUED
     priority: int = 0
-    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     heartbeat_at: datetime | None = None
 
     @field_validator("payload", mode="before")
@@ -61,7 +61,8 @@ def _decode_payload(cls, v: str | bytes) -> bytes:
                 return base64.b64decode(v)
             case _:
                 raise ValueError(
-                    f"payload must be bytes or a base64-encoded str, got {type(v).__name__}"
+                    "payload must be bytes or a base64-encoded str, "
+                    f"got {type(v).__name__}"
                 )
 
     @field_serializer("payload")
@@ -154,9 +155,7 @@ def with_job_removed(self, job_id: str) -> "QueueState":
         new_jobs = tuple(j for j in self.jobs if j.id != job_id)
         if len(new_jobs) == original_len:
             raise JobNotFoundError(job_id)
-        return self.model_copy(
-            update={"jobs": new_jobs, "version": self.version + 1}
-        )
+        return self.model_copy(update={"jobs": new_jobs, "version": self.version + 1})
 
     def requeue_stale(self, cutoff: datetime) -> "QueueState":
         """
@@ -175,6 +174,4 @@ def requeue_stale(self, cutoff: datetime) -> "QueueState":
         )
         if new_jobs == self.jobs:
             return self
-        return self.model_copy(
-            update={"jobs": new_jobs, "version": self.version + 1}
-        )
+        return self.model_copy(update={"jobs": new_jobs, "version": self.version + 1})

pyproject.toml

@@ -32,3 +32,11 @@ select = ["E", "F", "I", "UP"]
 [tool.mypy]
 strict = true
 python_version = "3.12"
+
+[[tool.mypy.overrides]]
+module = ["aioboto3", "aioboto3.*", "google.cloud.storage", "google.cloud.*", "google.api_core", "google.api_core.*"]
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+module = ["tests.*"]
+disallow_untyped_defs = false

tests/test_broker_queue.py

@@ -7,7 +7,6 @@
 from jqueue.domain.errors import JobNotFoundError
 from jqueue.domain.models import JobStatus
 
-
 # ---------------------------------------------------------------------------
 # Lifecycle
 # ---------------------------------------------------------------------------