fix: ruff 0.17 ICN001 and batch_bodies setter (ty 0.0.86)

faststream_outbox/_scheduling.py

@@ -11,27 +11,27 @@
 use ``resolve_next_attempt_client_side`` — only the batch and fake paths do.
 """
 
-import datetime as _dt
+import datetime as dt
 
 
 def is_future_dated(
-    activate_in: _dt.timedelta | None,
-    activate_at: _dt.datetime | None,
-    now: _dt.datetime,
+    activate_in: dt.timedelta | None,
+    activate_at: dt.datetime | None,
+    now: dt.datetime,
 ) -> bool:
     """Whether a row is genuinely future-dated (so NOTIFY is skipped — polling fires it at the gate)."""
     if activate_in is not None:
-        return activate_in > _dt.timedelta(0)
+        return activate_in > dt.timedelta(0)
     if activate_at is not None:
         return activate_at > now
     return False
 
 
 def resolve_next_attempt_client_side(
-    activate_in: _dt.timedelta | None,
-    activate_at: _dt.datetime | None,
-    now: _dt.datetime,
-) -> _dt.datetime | None:
+    activate_in: dt.timedelta | None,
+    activate_at: dt.datetime | None,
+    now: dt.datetime,
+) -> dt.datetime | None:
     """Resolve activate_in / activate_at to a single ``next_attempt_at`` value (client clock)."""
     if activate_in is not None:
         return now + activate_in
@@ -40,8 +40,8 @@ def resolve_next_attempt_client_side(
 
 def validate_activate_args(
     method_name: str,
-    activate_in: _dt.timedelta | None,
-    activate_at: _dt.datetime | None,
+    activate_in: dt.timedelta | None,
+    activate_at: dt.datetime | None,
 ) -> None:
     """Mutex + tz-aware checks shared by the test fakes. Real broker delegates to ``OutboxPublishCommand``."""
     if activate_in is not None and activate_at is not None:

faststream_outbox/_time.py

@@ -1,8 +1,8 @@
 """Shared time helper — single source so the five former ``_utcnow`` copies can't drift."""
 
-import datetime as _dt
+import datetime as dt
 
 
-def utcnow() -> _dt.datetime:
+def utcnow() -> dt.datetime:
     """Timezone-aware current UTC time."""
-    return _dt.datetime.now(tz=_dt.UTC)
+    return dt.datetime.now(tz=dt.UTC)

faststream_outbox/broker.py

@@ -6,7 +6,7 @@
 """
 
 import asyncio
-import datetime as _dt
+import datetime as dt
 import logging
 import typing
 import warnings
@@ -381,8 +381,8 @@ async def publish(  # ty: ignore[invalid-method-override]
         session: AsyncSession,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
-        activate_in: _dt.timedelta | None = None,
-        activate_at: _dt.datetime | None = None,
+        activate_in: dt.timedelta | None = None,
+        activate_at: dt.datetime | None = None,
         timer_id: str | None = None,
     ) -> int | None:
         """Insert one outbox row using *session*'s open transaction.
@@ -420,8 +420,8 @@ async def publish_batch(  # ty: ignore[invalid-method-override]
         queue: str,
         session: AsyncSession,
         headers: dict[str, str] | None = None,
-        activate_in: _dt.timedelta | None = None,
-        activate_at: _dt.datetime | None = None,
+        activate_in: dt.timedelta | None = None,
+        activate_at: dt.datetime | None = None,
     ) -> None:
         """Insert multiple outbox rows via *session*. Same transactional contract as ``publish``.
 

faststream_outbox/client.py

@@ -16,7 +16,7 @@
 
 import abc
 import asyncio
-import datetime as _dt
+import datetime as dt
 import uuid
 from typing import TYPE_CHECKING
 
@@ -117,8 +117,8 @@ async def mark_pending_with_lease(
         *,
         delay_seconds: float,
         attempts_count: int,
-        first_attempt_at: _dt.datetime,
-        last_attempt_at: _dt.datetime,
+        first_attempt_at: dt.datetime,
+        last_attempt_at: dt.datetime,
     ) -> bool: ...
 
     @abc.abstractmethod
@@ -398,8 +398,8 @@ async def mark_pending_with_lease(
         *,
         delay_seconds: float,
         attempts_count: int,
-        first_attempt_at: _dt.datetime,
-        last_attempt_at: _dt.datetime,
+        first_attempt_at: dt.datetime,
+        last_attempt_at: dt.datetime,
     ) -> bool:
         """Release the lease on *message_id* and reschedule it for retry, iff it still holds the lease.
 

faststream_outbox/message.py

@@ -8,7 +8,7 @@
 ``OutboxMessage`` adapts the inner message to FastStream's ``StreamMessage``.
 """
 
-import datetime as _dt
+import datetime as dt
 import logging
 import uuid
 from collections.abc import Awaitable, Callable
@@ -89,11 +89,11 @@ class OutboxInnerMessage:
     headers: dict[str, str] | None
     attempts_count: int
     deliveries_count: int
-    created_at: _dt.datetime
-    next_attempt_at: _dt.datetime
-    first_attempt_at: _dt.datetime | None
-    last_attempt_at: _dt.datetime | None
-    acquired_at: _dt.datetime | None
+    created_at: dt.datetime
+    next_attempt_at: dt.datetime
+    first_attempt_at: dt.datetime | None
+    last_attempt_at: dt.datetime | None
+    acquired_at: dt.datetime | None
     acquired_token: uuid.UUID | None
 
     # P9: the originating timer_id (single-publish dedup key), so a terminally-failed

faststream_outbox/publisher/usecase.py

@@ -8,7 +8,7 @@
 handler on the same session that owns the inbound row's terminal write.
 """
 
-import datetime as _dt
+import datetime as dt
 import typing
 
 from faststream._internal.endpoint.publisher import PublisherUsecase
@@ -71,8 +71,8 @@ async def publish(  # ty: ignore[invalid-method-override]
         session: AsyncSession,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
-        activate_in: _dt.timedelta | None = None,
-        activate_at: _dt.datetime | None = None,
+        activate_in: dt.timedelta | None = None,
+        activate_at: dt.datetime | None = None,
         timer_id: str | None = None,
     ) -> int | None:
         """Insert one outbox row scoped to this publisher's queue.

faststream_outbox/response.py

@@ -11,8 +11,9 @@
 ``from_cmd`` raises — see ``OutboxPublisher.__call__`` for the rationale.
 """
 
-import datetime as _dt
+import datetime as dt
 import typing
+from collections.abc import Sequence
 
 from faststream.response.publish_type import PublishType
 from faststream.response.response import BatchPublishCommand, PublishCommand, Response
@@ -32,8 +33,8 @@ def _validate_publish_args(
     *,
     queue: object,
     session: object,
-    activate_in: _dt.timedelta | None,
-    activate_at: _dt.datetime | None,
+    activate_in: dt.timedelta | None,
+    activate_at: dt.datetime | None,
 ) -> None:
     """Fail-fast validation shared by every real outbox publish entry point.
 
@@ -79,8 +80,8 @@ def __init__(
         session: AsyncSession,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
-        activate_in: _dt.timedelta | None = None,
-        activate_at: _dt.datetime | None = None,
+        activate_in: dt.timedelta | None = None,
+        activate_at: dt.datetime | None = None,
         timer_id: str | None = None,
         _publish_type: PublishType = PublishType.PUBLISH,
     ) -> None:
@@ -125,6 +126,11 @@ def batch_bodies(self) -> tuple[typing.Any, ...]:
         # OpenTelemetry batch-count attribute in one place.
         return (self.body, *self.extra_bodies)
 
+    @batch_bodies.setter
+    def batch_bodies(self, value: Sequence[typing.Any]) -> None:
+        self.body, *rest = value or (None,)
+        self.extra_bodies = tuple(rest)
+
     @classmethod
     def from_cmd(
         cls,
@@ -173,8 +179,8 @@ def __init__(
         session: AsyncSession,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
-        activate_in: _dt.timedelta | None = None,
-        activate_at: _dt.datetime | None = None,
+        activate_in: dt.timedelta | None = None,
+        activate_at: dt.datetime | None = None,
         timer_id: str | None = None,
     ) -> None:
         _validate_publish_args(

faststream_outbox/testing.py

@@ -11,7 +11,7 @@
 or scheduled delivery actually waiting. ``feed()`` simulates a row insert.
 """
 
-import datetime as _dt
+import datetime as dt
 import typing
 import uuid
 from contextlib import contextmanager
@@ -47,16 +47,16 @@ class _FakeRow:
     headers: dict[str, str] | None
     attempts_count: int = 0
     deliveries_count: int = 0
-    created_at: _dt.datetime = field(default_factory=utcnow)
-    next_attempt_at: _dt.datetime = field(default_factory=utcnow)
-    first_attempt_at: _dt.datetime | None = None
-    last_attempt_at: _dt.datetime | None = None
-    acquired_at: _dt.datetime | None = None
+    created_at: dt.datetime = field(default_factory=utcnow)
+    next_attempt_at: dt.datetime = field(default_factory=utcnow)
+    first_attempt_at: dt.datetime | None = None
+    last_attempt_at: dt.datetime | None = None
+    acquired_at: dt.datetime | None = None
     acquired_token: uuid.UUID | None = None
     timer_id: str | None = None
 
 
-def _claim_fake_row(row: _FakeRow, *, now: _dt.datetime, token: uuid.UUID) -> None:
+def _claim_fake_row(row: _FakeRow, *, now: dt.datetime, token: uuid.UUID) -> None:
     """Set the lease and bump ``deliveries_count`` — the shared claim mechanics (F1-08).
 
     Both ``FakeOutboxClient.fetch`` (loop mode) and ``_sync_dispatch`` (sync mode) route
@@ -99,7 +99,7 @@ def feed(
         queue: str,
         payload: bytes,
         headers: dict[str, str] | None = None,
-        next_attempt_at: _dt.datetime | None = None,
+        next_attempt_at: dt.datetime | None = None,
         timer_id: str | None = None,
     ) -> int | None:
         # P31: reject naive datetimes up front — the publish path is tz-strict, and a
@@ -156,7 +156,7 @@ async def fetch(
         if not queues:
             return []
         now = utcnow()
-        lease_cutoff = now - _dt.timedelta(seconds=max(0.0, lease_ttl_seconds))
+        lease_cutoff = now - dt.timedelta(seconds=max(0.0, lease_ttl_seconds))
         token = uuid.uuid4()
         out: list[OutboxInnerMessage] = []
         eligible = sorted(
@@ -226,12 +226,12 @@ async def mark_pending_with_lease(
         *,
         delay_seconds: float,
         attempts_count: int,
-        first_attempt_at: _dt.datetime,
-        last_attempt_at: _dt.datetime,
+        first_attempt_at: dt.datetime,
+        last_attempt_at: dt.datetime,
     ) -> bool:
         for row in self._rows:
             if row.id == message_id and _lease_token_matches(row.acquired_token, acquired_token):
-                row.next_attempt_at = utcnow() + _dt.timedelta(seconds=max(0.0, delay_seconds))
+                row.next_attempt_at = utcnow() + dt.timedelta(seconds=max(0.0, delay_seconds))
                 row.attempts_count = attempts_count
                 row.first_attempt_at = first_attempt_at
                 row.last_attempt_at = last_attempt_at
@@ -371,7 +371,7 @@ async def _fake_publish_one(
     queue: str,
     headers: dict[str, str] | None,
     correlation_id: str | None,
-    next_at: "_dt.datetime | None",
+    next_at: "dt.datetime | None",
     timer_id: str | None,
     run_loops: bool,
 ) -> int | None:
@@ -396,7 +396,7 @@ async def _fake_publish_many(
     bodies: "Sequence[typing.Any]",
     queue: str,
     headers: dict[str, str] | None,
-    next_at: "_dt.datetime | None",
+    next_at: "dt.datetime | None",
     run_loops: bool,
 ) -> None:
     """Shared batch insert for both batch paths (P29).
@@ -507,8 +507,8 @@ async def fake_publish(
         session: typing.Any = None,
         headers: dict[str, str] | None = None,
         correlation_id: str | None = None,
-        activate_in: _dt.timedelta | None = None,
-        activate_at: _dt.datetime | None = None,
+        activate_in: dt.timedelta | None = None,
+        activate_at: dt.datetime | None = None,
         timer_id: str | None = None,
     ) -> int | None:
         # P33: test mode is intentionally lenient about ``session`` — the fake client has
@@ -546,8 +546,8 @@ async def fake_publish_batch(
         queue: str,
         session: typing.Any = None,
         headers: dict[str, str] | None = None,
-        activate_in: _dt.timedelta | None = None,
-        activate_at: _dt.datetime | None = None,
+        activate_in: dt.timedelta | None = None,
+        activate_at: dt.datetime | None = None,
     ) -> None:
         del session
         validate_activate_args("broker.publish_batch", activate_in, activate_at)
@@ -604,7 +604,7 @@ def feed(
         payload: bytes,
         *,
         headers: dict[str, str] | None = None,
-        next_attempt_at: _dt.datetime | None = None,
+        next_attempt_at: dt.datetime | None = None,
         timer_id: str | None = None,
     ) -> int | None:
         """Insert a row directly into the in-memory store. Returns the row id, or None on timer_id conflict."""

tests/test_client_contract.py

@@ -25,7 +25,7 @@
 authority stays a documented invariant (AGENTS.md), not an assertion here.
 """
 
-import datetime as _dt
+import datetime as dt
 import os
 import uuid
 from collections.abc import AsyncIterator
@@ -68,15 +68,15 @@ async def seed(
         rid = self._client.feed(
             queue=queue,
             payload=payload,
-            next_attempt_at=utcnow() + _dt.timedelta(seconds=next_attempt_offset),
+            next_attempt_at=utcnow() + dt.timedelta(seconds=next_attempt_offset),
             timer_id=timer_id,
         )
         assert rid is not None
         row = next(r for r in self._client.rows if r.id == rid)
         row.deliveries_count = deliveries_count
         if leased_token is not None:
             row.acquired_token = leased_token
-            row.acquired_at = utcnow() - _dt.timedelta(seconds=acquired_age or 0.0)
+            row.acquired_at = utcnow() - dt.timedelta(seconds=acquired_age or 0.0)
         return rid
 
     async def fetch(self, queues: list[str], *, limit: int, lease_ttl_seconds: float) -> list[OutboxInnerMessage]:
@@ -101,8 +101,8 @@ async def mark_pending(
         *,
         delay_seconds: float,
         attempts_count: int,
-        first_attempt_at: _dt.datetime,
-        last_attempt_at: _dt.datetime,
+        first_attempt_at: dt.datetime,
+        last_attempt_at: dt.datetime,
     ) -> bool:
         return await self._client.mark_pending_with_lease(
             None,
@@ -200,8 +200,8 @@ async def mark_pending(
         *,
         delay_seconds: float,
         attempts_count: int,
-        first_attempt_at: _dt.datetime,
-        last_attempt_at: _dt.datetime,
+        first_attempt_at: dt.datetime,
+        last_attempt_at: dt.datetime,
     ) -> bool:
         async with self._engine.connect() as raw:
             writer = await raw.execution_options(isolation_level="AUTOCOMMIT")
@@ -449,7 +449,7 @@ async def test_mark_pending_reschedules_on_token_match(contract: _Harness) -> No
     assert row["acquired_at"] is None
     assert row["attempts_count"] == 1
     next_attempt_at = row["next_attempt_at"]
-    assert isinstance(next_attempt_at, _dt.datetime)
+    assert isinstance(next_attempt_at, dt.datetime)
     assert next_attempt_at > utcnow()