Repository: airtai/faststream
Pull request #1930 (2024-11-20 19:50:18)

AsyncAPI 2.6.0 fix empty channels for KafkaSubscriber and ConfluentSubscriber if partitions provided
…rovided

# Description

Please include a summary of the change and specify which issue is being addressed. Additionally, provide relevant motivation and context.

Fixes # (issue number)

## Type of change

Please delete options that are not relevant.

- [ ] Documentation (typos, code examples, or any documentation updates)
- [x] Bug fix (a non-breaking change that resolves an issue)
- [ ] New feature (a non-breaking change that adds functionality)
- [ ] Breaking change (a fix or feature that would disrupt existing functionality)
- [ ] This change requires a documentation update

## Checklist

- [ ] My code adheres to the style guidelines of this project (`scripts/lint.sh` shows no errors)
- [ ] I have conducted a self-review of my own code
- [ ] I have made the necessary changes to the documentation
- [ ] My changes do not generate any new warnings
- [ ] I have added tests to validate the effectiveness of my fix or the functionality of my new feature
- [ ] Both new and existing unit tests pass successfully on my local environment by running `scripts/test-cov.sh`
- [ ] I have ensured that static analysis tests are passing by running `scripts/static-analysis.sh`
- [ ] I have included code examples to illustrate the modifications



Patch:
diff --git a/faststream/confluent/subscriber/asyncapi.py b/faststream/confluent/subscriber/asyncapi.py
index ca56bec6..ee87c0e0 100644
--- a/faststream/confluent/subscriber/asyncapi.py
+++ b/faststream/confluent/subscriber/asyncapi.py
@@ -1,3 +1,4 @@
+from itertools import chain
 from typing import (
     TYPE_CHECKING,
     Dict,
@@ -34,7 +35,10 @@ class AsyncAPISubscriber(LogicSubscriber[MsgType]):
         channels = {}
 
         payloads = self.get_payloads()
-        for t in self.topics:
+
+        topics = chain(self.topics, {part.topic for part in self.partitions})
+
+        for t in topics:
             handler_name = self.title_ or f"{t}:{self.call_name}"
 
             channels[handler_name] = Channel(
diff --git a/faststream/kafka/subscriber/asyncapi.py b/faststream/kafka/subscriber/asyncapi.py
index 9adb8dad..87fff1c2 100644
--- a/faststream/kafka/subscriber/asyncapi.py
+++ b/faststream/kafka/subscriber/asyncapi.py
@@ -1,3 +1,4 @@
+from itertools import chain
 from typing import (
     TYPE_CHECKING,
     Dict,
@@ -35,7 +36,9 @@ class AsyncAPISubscriber(LogicSubscriber[MsgType]):
 
         payloads = self.get_payloads()
 
-        for t in self.topics:
+        topics = chain(self.topics, {part.topic for part in self.partitions})
+
+        for t in topics:
             handler_name = self.title_ or f"{t}:{self.call_name}"
 
             channels[handler_name] = Channel(
diff --git a/faststream/nats/fastapi/fastapi.py b/faststream/nats/fastapi/fastapi.py
index 4a871426..6701ef36 100644
--- a/faststream/nats/fastapi/fastapi.py
+++ b/faststream/nats/fastapi/fastapi.py
@@ -203,6 +203,10 @@ class NatsRouter(StreamRouter["Msg"]):
             Optional[str],
             Doc("Nkeys seed to be used."),
         ] = None,
+        nkeys_seed_str: Annotated[
+            Optional[str],
+            Doc("Raw nkeys seed to be used."),
+        ] = None,
         inbox_prefix: Annotated[
             Union[str, bytes],
             Doc(
@@ -533,6 +537,7 @@ class NatsRouter(StreamRouter["Msg"]):
             user_jwt_cb=user_jwt_cb,
             user_credentials=user_credentials,
             nkeys_seed=nkeys_seed,
+            nkeys_seed_str=nkeys_seed_str,
             inbox_prefix=inbox_prefix,
             pending_size=pending_size,
             flush_timeout=flush_timeout,
diff --git a/faststream/prometheus/middleware.py b/faststream/prometheus/middleware.py
index 575d8463..aabe7028 100644
--- a/faststream/prometheus/middleware.py
+++ b/faststream/prometheus/middleware.py
@@ -2,6 +2,7 @@ import time
 from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence
 
 from faststream import BaseMiddleware
+from faststream.exceptions import IgnoredException
 from faststream.prometheus.consts import (
     PROCESSING_STATUS_BY_ACK_STATUS,
     PROCESSING_STATUS_BY_HANDLER_EXCEPTION_MAP,
@@ -71,11 +72,13 @@ class PrometheusMiddleware(BaseMiddleware):
 
         except Exception as e:
             err = e
-            self._metrics_manager.add_received_processed_message_exception(
-                exception_type=type(err).__name__,
-                broker=messaging_system,
-                handler=destination_name,
-            )
+
+            if not isinstance(err, IgnoredException):
+                self._metrics_manager.add_received_processed_message_exception(
+                    exception_type=type(err).__name__,
+                    broker=messaging_system,
+                    handler=destination_name,
+                )
             raise
 
         finally:
diff --git a/faststream/rabbit/broker/broker.py b/faststream/rabbit/broker/broker.py
index 774a8866..01ae9d6b 100644
--- a/faststream/rabbit/broker/broker.py
+++ b/faststream/rabbit/broker/broker.py
@@ -839,7 +839,7 @@ class RabbitBroker(
                 if cancel_scope.cancel_called:
                     return False
 
-                if not self._connection.is_closed:
+                if self._connection.connected.is_set():
                     return True
 
                 await anyio.sleep(sleep_time)


Test patch:
diff --git a/tests/asyncapi/confluent/test_arguments.py b/tests/asyncapi/confluent/test_arguments.py
index 9e3746e3..2eb5c91c 100644
--- a/tests/asyncapi/confluent/test_arguments.py
+++ b/tests/asyncapi/confluent/test_arguments.py
@@ -1,5 +1,5 @@
 from faststream.asyncapi.generate import get_app_schema
-from faststream.confluent import KafkaBroker
+from faststream.confluent import KafkaBroker, TopicPartition
 from tests.asyncapi.base.arguments import ArgumentsTestcase
 
 
@@ -18,3 +18,39 @@ class TestArguments(ArgumentsTestcase):
         assert schema["channels"][key]["bindings"] == {
             "kafka": {"bindingVersion": "0.4.0", "topic": "test"}
         }
+
+    def test_subscriber_with_one_topic_partitions(self):
+        broker = self.broker_class()
+
+        part1 = TopicPartition("topic_name", 1)
+        part2 = TopicPartition("topic_name", 2)
+
+        @broker.subscriber(partitions=[part1, part2])
+        async def handle(msg): ...
+
+        schema = get_app_schema(self.build_app(broker)).to_jsonable()
+        key = tuple(schema["channels"].keys())[0]  # noqa: RUF015
+
+        assert schema["channels"][key]["bindings"] == {
+            "kafka": {"bindingVersion": "0.4.0", "topic": "topic_name"}
+        }
+
+    def test_subscriber_with_multi_topics_partitions(self):
+        broker = self.broker_class()
+
+        part1 = TopicPartition("topic_name1", 1)
+        part2 = TopicPartition("topic_name2", 2)
+
+        @broker.subscriber(partitions=[part1, part2])
+        async def handle(msg): ...
+
+        schema = get_app_schema(self.build_app(broker)).to_jsonable()
+        key1 = tuple(schema["channels"].keys())[0]  # noqa: RUF015
+        key2 = tuple(schema["channels"].keys())[1]
+
+        assert sorted(
+            (
+                schema["channels"][key1]["bindings"]["kafka"]["topic"],
+                schema["channels"][key2]["bindings"]["kafka"]["topic"],
+            )
+        ) == sorted(("topic_name1", "topic_name2"))
diff --git a/tests/asyncapi/kafka/test_arguments.py b/tests/asyncapi/kafka/test_arguments.py
index 89595c06..5b289ffd 100644
--- a/tests/asyncapi/kafka/test_arguments.py
+++ b/tests/asyncapi/kafka/test_arguments.py
@@ -1,3 +1,5 @@
+from aiokafka import TopicPartition
+
 from faststream.asyncapi.generate import get_app_schema
 from faststream.kafka import KafkaBroker
 from tests.asyncapi.base.arguments import ArgumentsTestcase
@@ -18,3 +20,39 @@ class TestArguments(ArgumentsTestcase):
         assert schema["channels"][key]["bindings"] == {
             "kafka": {"bindingVersion": "0.4.0", "topic": "test"}
         }
+
+    def test_subscriber_with_one_topic_partitions(self):
+        broker = self.broker_class()
+
+        part1 = TopicPartition("topic_name", 1)
+        part2 = TopicPartition("topic_name", 2)
+
+        @broker.subscriber(partitions=[part1, part2])
+        async def handle(msg): ...
+
+        schema = get_app_schema(self.build_app(broker)).to_jsonable()
+        key = tuple(schema["channels"].keys())[0]  # noqa: RUF015
+
+        assert schema["channels"][key]["bindings"] == {
+            "kafka": {"bindingVersion": "0.4.0", "topic": "topic_name"}
+        }
+
+    def test_subscriber_with_multi_topics_partitions(self):
+        broker = self.broker_class()
+
+        part1 = TopicPartition("topic_name1", 1)
+        part2 = TopicPartition("topic_name2", 2)
+
+        @broker.subscriber(partitions=[part1, part2])
+        async def handle(msg): ...
+
+        schema = get_app_schema(self.build_app(broker)).to_jsonable()
+        key1 = tuple(schema["channels"].keys())[0]  # noqa: RUF015
+        key2 = tuple(schema["channels"].keys())[1]
+
+        assert sorted(
+            (
+                schema["channels"][key1]["bindings"]["kafka"]["topic"],
+                schema["channels"][key2]["bindings"]["kafka"]["topic"],
+            )
+        ) == sorted(("topic_name1", "topic_name2"))
diff --git a/tests/prometheus/basic.py b/tests/prometheus/basic.py
index 1d749af1..9f80a86b 100644
--- a/tests/prometheus/basic.py
+++ b/tests/prometheus/basic.py
@@ -7,7 +7,7 @@ from prometheus_client import CollectorRegistry
 
 from faststream import Context
 from faststream.broker.message import AckStatus
-from faststream.exceptions import RejectMessage
+from faststream.exceptions import IgnoredException, RejectMessage
 from faststream.prometheus.middleware import (
     PROCESSING_STATUS_BY_ACK_STATUS,
     PROCESSING_STATUS_BY_HANDLER_EXCEPTION_MAP,
@@ -54,6 +54,11 @@ class LocalPrometheusTestcase(BaseTestcaseConfig):
             pytest.param(
                 AckStatus.rejected, None, id="rejected status without exception"
             ),
+            pytest.param(
+                AckStatus.acked,
+                IgnoredException,
+                id="acked status with ignore exception",
+            ),
         ],
     )
     async def test_metrics(
@@ -176,7 +181,7 @@ class LocalPrometheusTestcase(BaseTestcaseConfig):
             ),
         ]
 
-        if status == ProcessingStatus.error:
+        if exception_class and not issubclass(exception_class, IgnoredException):
             assert (
                 metrics_manager.add_received_processed_message_exception.mock_calls
                 == [
@@ -187,6 +192,11 @@ class LocalPrometheusTestcase(BaseTestcaseConfig):
                     ),
                 ]
             )
+        else:
+            assert (
+                metrics_manager.add_received_processed_message_exception.mock_calls
+                == []
+            )
 
     def assert_publish_metrics(self, metrics_manager: Any):
         settings_provider = self.settings_provider_factory(None)