fix: P27 — attribute subscriber-config warnings to the user's call site

faststream_outbox/subscriber/factory.py

@@ -1,6 +1,8 @@
 import typing
 import warnings
+from pathlib import Path
 
+import faststream
 from faststream._internal.constants import EMPTY
 from faststream._internal.endpoint.subscriber.call_item import CallsCollection
 from faststream.middlewares import AckPolicy
@@ -15,6 +17,16 @@
     from faststream_outbox.retry import RetryStrategyProto
 
 
+# P27: attribute the subscriber-config warnings to the user's ``@broker.subscriber(...)``
+# / ``@router.subscriber(...)`` call site by skipping frames inside this package and
+# faststream. A static ``stacklevel`` can't be correct for both the direct path and the
+# FastAPI-router path (the router adds frames); ``skip_file_prefixes`` (3.12+) is.
+_WARN_SKIP_PREFIXES = (
+    str(Path(__file__).parent.parent),  # the faststream_outbox package dir
+    str(Path(faststream.__file__).parent),  # the faststream package dir
+)
+
+
 def create_subscriber(
     *,
     queues: list[str],
@@ -89,9 +101,9 @@ def _validate_subscriber_config(  # noqa: C901  # flat sequence of independent k
     Reject impossible knob values, warn on combos that silently misbehave.
 
     Errors are raised here (not deferred to runtime) so the user gets a
-    traceback pointing at the ``@broker.subscriber(...)`` decorator. Warnings
-    use ``stacklevel=4`` so they point at the same line (frames: user →
-    ``subscriber()`` → ``create_subscriber()`` → ``_validate_subscriber_config()``).
+    traceback pointing at the ``@broker.subscriber(...)`` decorator. Warnings use
+    ``skip_file_prefixes`` (see ``_WARN_SKIP_PREFIXES``) so they are attributed to the
+    user's call site on both the direct and FastAPI-router paths (P27).
     """
     if max_workers <= 0:
         msg = f"max_workers must be >= 1, got {max_workers}"
@@ -133,22 +145,22 @@ def _validate_subscriber_config(  # noqa: C901  # flat sequence of independent k
             "retry_strategy is ignored. Pass ack_policy=NACK_ON_ERROR (default) to "
             "honor retry, or drop retry_strategy if you really want first-error deletion.",
             UserWarning,
-            stacklevel=4,
+            skip_file_prefixes=_WARN_SKIP_PREFIXES,
         )
     if ack_policy is AckPolicy.NACK_ON_ERROR and is_no_retry:
         warnings.warn(
             "ack_policy=NACK_ON_ERROR with retry_strategy=NoRetry() has the same effect "
             "as REJECT_ON_ERROR (one attempt, then delete). Pick one for clarity.",
             UserWarning,
-            stacklevel=4,
+            skip_file_prefixes=_WARN_SKIP_PREFIXES,
         )
     if max_deliveries is not None and (retry_strategy is None or is_no_retry):
         warnings.warn(
             "max_deliveries is set but no retry_strategy is configured (or NoRetry was "
             "passed); the delivery cap is unreachable on the happy path since the row "
             "is deleted after the first attempt.",
             UserWarning,
-            stacklevel=4,
+            skip_file_prefixes=_WARN_SKIP_PREFIXES,
         )
     if lease_ttl_seconds <= max_fetch_interval:
         warnings.warn(
@@ -158,5 +170,5 @@ def _validate_subscriber_config(  # noqa: C901  # flat sequence of independent k
             f"of healthy in-flight rows. Recommended: lease_ttl_seconds >= "
             f"2 * max_fetch_interval + P99(handler).",
             UserWarning,
-            stacklevel=4,
+            skip_file_prefixes=_WARN_SKIP_PREFIXES,
         )

tests/test_fastapi.py

@@ -17,10 +17,11 @@
 import pytest
 from fastapi import Depends, FastAPI
 from fastapi.testclient import TestClient
+from faststream.middlewares import AckPolicy
 from sqlalchemy import MetaData
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from faststream_outbox import OutboxMessage, make_outbox_table
+from faststream_outbox import NoRetry, OutboxMessage, make_outbox_table
 from faststream_outbox.client import AbstractOutboxClient
 from faststream_outbox.fastapi import (
     OutboxBroker as AnnotatedOutboxBroker,
@@ -77,6 +78,22 @@ async def handle(body: dict) -> None:
     assert received == [{"x": 1}]
 
 
+def test_subscriber_misconfig_warning_attributed_to_user_via_fastapi_router() -> None:
+    """
+    P27: through the FastAPI router (extra frames) the misconfig warning points at the user's call site.
+
+    Old static ``stacklevel=4`` landed on a faststream-internal frame on this path; the
+    ``skip_file_prefixes`` attribution lands on the user's ``@router.subscriber(...)`` line.
+    """
+    router = OutboxRouter(outbox_table=_make_outbox_table())
+    with pytest.warns(UserWarning, match="NACK_ON_ERROR") as record:
+
+        @router.subscriber("orders", ack_policy=AckPolicy.NACK_ON_ERROR, retry_strategy=NoRetry())
+        async def handle(body: dict) -> None: ...
+
+    assert record[0].filename == __file__  # attributed to this test (the user), not a package frame
+
+
 async def test_subscriber_receives_fastapi_depends_session() -> None:
     """B-load-bearing: ``Depends(get_session)`` resolves inside an outbox subscriber handler."""
     t = _make_outbox_table()