.

rabbit_base.py

@@ -6,6 +6,22 @@ import ssl
 
 ExchangeResolver = Callable[[str], str]  # exchangeName -> exchangeType
 
+# rabbit_base.py
+import aio_pika
+
+def _normalize_exchange_type(val: str) -> aio_pika.ExchangeType:
+    # 1) Try attribute by NAME (DIRECT/FANOUT/TOPIC/HEADERS)
+    if isinstance(val, str):
+        name = val.upper()
+        if hasattr(aio_pika.ExchangeType, name):
+            return getattr(aio_pika.ExchangeType, name)
+        # 2) Try enum by VALUE ("direct"/"fanout"/"topic"/"headers")
+        try:
+            return aio_pika.ExchangeType(val.lower())
+        except Exception:
+            pass
+    # 3) Default
+    return aio_pika.ExchangeType.TOPIC
 
 def _parse_amqp_url(url: str) -> dict:
     parts = urlsplit(url)
@@ -53,7 +69,9 @@ class RabbitBase:
         await self.connect()
         if name in self._exchanges:
             return self._exchanges[name]
-        ex_type = self._exchange_type_resolver(name)
+        ex_type_str = self._exchange_type_resolver(name)  # e.g. "direct"
+        ex_type = _normalize_exchange_type(ex_type_str)
+
         ex = await self._chan.declare_exchange(
             name, getattr(aio_pika.ExchangeType, ex_type), durable=True
         )