Bufixes 745

faststream/__about__.py

@@ -1,5 +1,5 @@
 """Simple and fast framework to create message brokers based microservices"""
-__version__ = "0.1.3"
+__version__ = "0.1.4"
 
 
 INSTALL_YAML = """

faststream/asyncapi/schema/info.py

@@ -9,8 +9,8 @@
     JsonSchemaValue,
     Required,
     TypedDict,
-    with_info_plain_validator_function,
     is_installed,
+    with_info_plain_validator_function,
 )
 from faststream.log import logger
 

faststream/broker/fastapi/route.py

@@ -232,8 +232,8 @@ async def app(message: NativeMessage[Any]) -> SendableMessage:
             """
             body = message.decoded_body
             if first_arg is not None:
-                if not isinstance(body, dict):  # pragma: no branch
-                    fastapi_body: AnyDict = {first_arg: body}
+                if not isinstance(body, dict) and not isinstance(body, list):
+                    fastapi_body: Any = {first_arg: body}
                 else:
                     fastapi_body = body
 

faststream/kafka/broker.py

@@ -374,7 +374,7 @@ def subscriber(  # type: ignore[override]
 
         self._setup_log_context(topics)
 
-        key = "".join(topics)
+        key = Handler.get_routing_hash(topics, group_id)
         builder = partial(
             aiokafka.AIOKafkaConsumer,
             key_deserializer=key_deserializer,

faststream/kafka/handler.py

@@ -226,3 +226,7 @@ async def _consume(self) -> Never:
                 if connected is False:
                     connected = True
                 await self.consume(msg)
+
+    @staticmethod
+    def get_routing_hash(topics: Sequence[str], group_id: Optional[str] = None) -> str:
+        return "".join((*topics, group_id or ""))