fix: support all RMQ exchanges in AsyncAPI

faststream/asyncapi/schema/bindings/amqp.py

@@ -37,8 +37,18 @@ class Exchange(BaseModel):
         vhost : virtual host of the exchange, default is "/"
     """
 
+    type: Literal[
+        "default",
+        "direct",
+        "topic",
+        "fanout",
+        "headers",
+        "x-delayed-message",
+        "x-consistent-hash",
+        "x-modulus-hash",
+    ]
+
     name: Optional[str] = None
-    type: Literal["default", "direct", "topic", "fanout", "headers"]
     durable: Optional[bool] = None
     autoDelete: Optional[bool] = None
     vhost: str = "/"

pyproject.toml

@@ -111,9 +111,9 @@ types = [
 
 lint = [
     "faststream[types]",
-    "ruff==0.5.6",
+    "ruff==0.5.7",
     "bandit==1.7.9",
-    "semgrep==1.84.0",
+    "semgrep==1.84.1",
     "codespell==2.3.0",
 ]
 
@@ -130,8 +130,8 @@ testing = [
     "fastapi==0.112.0",
     "pydantic-settings>=2.0.0,<3.0.0",
     "httpx==0.27.0",
-    "PyYAML==6.0.1",
-    "watchfiles==0.22.0",
+    "PyYAML==6.0.2",
+    "watchfiles==0.23.0",
     "email-validator==2.2.0",
 ]
 

tests/asyncapi/rabbit/test_arguments.py

@@ -64,3 +64,107 @@ async def handle(msg): ...
                 "is": "routingKey",
             }
         }
+
+    def test_subscriber_headers_bindings(self):
+        broker = self.broker_class()
+
+        @broker.subscriber(
+            RabbitQueue("test", auto_delete=True),
+            RabbitExchange("test-ex", type=ExchangeType.HEADERS),
+        )
+        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"] == {
+            "amqp": {
+                "bindingVersion": "0.2.0",
+                "exchange": {
+                    "autoDelete": False,
+                    "durable": False,
+                    "name": "test-ex",
+                    "type": "headers",
+                    "vhost": "/",
+                },
+                "is": "routingKey",
+            }
+        }
+
+    def test_subscriber_xdelay_bindings(self):
+        broker = self.broker_class()
+
+        @broker.subscriber(
+            RabbitQueue("test", auto_delete=True),
+            RabbitExchange("test-ex", type=ExchangeType.X_DELAYED_MESSAGE),
+        )
+        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"] == {
+            "amqp": {
+                "bindingVersion": "0.2.0",
+                "exchange": {
+                    "autoDelete": False,
+                    "durable": False,
+                    "name": "test-ex",
+                    "type": "x-delayed-message",
+                    "vhost": "/",
+                },
+                "is": "routingKey",
+            }
+        }
+
+    def test_subscriber_consistent_hash_bindings(self):
+        broker = self.broker_class()
+
+        @broker.subscriber(
+            RabbitQueue("test", auto_delete=True),
+            RabbitExchange("test-ex", type=ExchangeType.X_CONSISTENT_HASH),
+        )
+        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"] == {
+            "amqp": {
+                "bindingVersion": "0.2.0",
+                "exchange": {
+                    "autoDelete": False,
+                    "durable": False,
+                    "name": "test-ex",
+                    "type": "x-consistent-hash",
+                    "vhost": "/",
+                },
+                "is": "routingKey",
+            }
+        }
+
+    def test_subscriber_modules_hash_bindings(self):
+        broker = self.broker_class()
+
+        @broker.subscriber(
+            RabbitQueue("test", auto_delete=True),
+            RabbitExchange("test-ex", type=ExchangeType.X_MODULUS_HASH),
+        )
+        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"] == {
+            "amqp": {
+                "bindingVersion": "0.2.0",
+                "exchange": {
+                    "autoDelete": False,
+                    "durable": False,
+                    "name": "test-ex",
+                    "type": "x-modulus-hash",
+                    "vhost": "/",
+                },
+                "is": "routingKey",
+            }
+        }