diff --git a/CLAUDE.md b/CLAUDE.md index 357e358..2847fb0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -8,7 +8,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co ## Commands -`just` (task runner) + `uv` (package manager); the [`Justfile`](Justfile) is the source of truth for recipes — run `just --list` or read it. The non-obvious bits: +`just` (task runner) + `uv` (package manager); the [`justfile`](justfile) is the source of truth for recipes — run `just --list` or read it. The non-obvious bits: - `just test [args]` — full suite in docker compose (Postgres 17). Args forward **unquoted**, so a spaced `-k` expression (`-k "a or b"`) word-splits and fails (`file or directory not found: or`) — run one keyword per invocation, or a single substring matching all targets. `tests/test_unit.py` + `tests/test_fake.py` need no Postgres (`uv run pytest tests/test_unit.py` works directly); `tests/test_integration.py` needs Postgres at `POSTGRES_DSN` (default `postgresql+asyncpg://outbox:outbox@localhost:5432/outbox`; `pg_engine` skips if unreachable). Coverage is on with `--cov-fail-under=100` — partial runs fail that gate; pass `--no-cov` or `--cov-fail-under=0` when iterating. - `just lint` / `just lint-ci` — autofix vs non-mutating; `lint-ci` also runs the planning-change validator. diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py index 65447cb..7b04876 100644 --- a/faststream_outbox/subscriber/usecase.py +++ b/faststream_outbox/subscriber/usecase.py @@ -49,7 +49,7 @@ try: import asyncpg as _asyncpg except ImportError: # pragma: no cover - _asyncpg = None # ty: ignore[invalid-assignment] + _asyncpg = None _BACKOFF_EXP_CAP = 30 @@ -145,6 +145,9 @@ def _render_last_exception( if typing.TYPE_CHECKING: + # Type-only handle on the module: `_asyncpg` above is `module | None`, so annotations + # must not hang off it (attribute access on the `None` arm is unresolved). + import asyncpg from faststream._internal.endpoint.publisher import PublisherProto from faststream._internal.endpoint.subscriber.call_item import CallsCollection from faststream.message import StreamMessage @@ -356,7 +359,7 @@ async def _fetch_inner( self, *, fetch_conn: "AsyncConnection | None", - listen_conn: "_asyncpg.Connection | None", + listen_conn: "asyncpg.Connection | None", ) -> None: """Fetch + adaptive backoff, with NOTIFY-driven wakeup. @@ -415,7 +418,7 @@ async def _wait_for_notify_or_timeout(self, timeout: float) -> None: # noqa: AS await asyncio.wait_for(self._notify_event.wait(), timeout=timeout) self._notify_event.clear() - async def _open_listen_connection(self, engine: "AsyncEngine") -> "_asyncpg.Connection | None": + async def _open_listen_connection(self, engine: "AsyncEngine") -> "asyncpg.Connection | None": """Open a dedicated raw asyncpg connection and register LISTEN on it. Returns the connection on success, ``None`` on any failure (asyncpg not installed, @@ -437,7 +440,7 @@ async def _open_listen_connection(self, engine: "AsyncEngine") -> "_asyncpg.Conn _, opts = engine.dialect.create_connect_args(engine.url) for sa_only_key in ("prepared_statement_cache_size", "async_fallback", "async_creator_fn"): opts.pop(sa_only_key, None) - conn: _asyncpg.Connection | None = None + conn: asyncpg.Connection | None = None listening = False try: conn = await _asyncpg.connect(**opts) @@ -459,7 +462,7 @@ async def _open_listen_connection(self, engine: "AsyncEngine") -> "_asyncpg.Conn await conn.close() return conn if listening else None - async def _close_listen_connection(self, listen_conn: "_asyncpg.Connection") -> None: + async def _close_listen_connection(self, listen_conn: "asyncpg.Connection") -> None: """Close the raw LISTEN connection without letting teardown wedge the fetch loop (S1). A graceful ``close()`` on a half-dead socket can block on the kernel keepalive diff --git a/Justfile b/justfile similarity index 100% rename from Justfile rename to justfile