Refectored Code slightly

airtai · May 18, 2024 · fff8a92 · fff8a92
1 parent e77be6c
commit fff8a92
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 25 deletions.
diff --git a/faststream/confluent/testing.py b/faststream/confluent/testing.py
@@ -109,21 +109,10 @@ async def publish(  # type: ignore[override]
         return_value = None
 
         for handler in self.broker._subscribers.values():  # pragma: no branch
-            call = False
-
-            # for p in handler.partitions:
-            #     if p.topic == topic and (partition is None or p.partition == partition):
-            #         call = True
-
             if topic in handler.topics:
-                call = True
-
-            if call:
                 handle_value = await call_handler(
                     handler=handler,
-                    message=[incoming]
-                    if isinstance(handler, AsyncAPIBatchSubscriber)
-                    else incoming,
+                    message=[incoming] if isinstance(handler, AsyncAPIBatchSubscriber) else incoming,
                     rpc=rpc,
                     rpc_timeout=rpc_timeout,
                     raise_timeout=raise_timeout,

diff --git a/faststream/kafka/testing.py b/faststream/kafka/testing.py
@@ -108,21 +108,13 @@ async def publish(  # type: ignore[override]
         return_value = None
 
         for handler in self.broker._subscribers.values():  # pragma: no branch
-            call = False
-
-            for p in handler.partitions:
-                if p.topic == topic and (partition is None or p.partition == partition):
-                    call = True
-
-            if topic in handler.topics:
-                call = True
-
-            if call:
+            if any(
+                    p.topic == topic and (partition is None or p.partition == partition)
+                    for p in handler.partitions
+            ) or topic in handler.topics:
                 handle_value = await call_handler(
                     handler=handler,
-                    message=[incoming]
-                    if isinstance(handler, AsyncAPIBatchSubscriber)
-                    else incoming,
+                    message=[incoming] if isinstance(handler, AsyncAPIBatchSubscriber) else incoming,
                     rpc=rpc,
                     rpc_timeout=rpc_timeout,
                     raise_timeout=raise_timeout,