.

listener.py

@@ -1,11 +1,16 @@
+# listener.py
 import json
+from typing import Callable, Awaitable, Dict, Any, List
+
 import aio_pika
-from typing import Callable, Awaitable, Dict, Any, List, Optional
+
+Handler = Callable[[Any], Awaitable[None]]  # payload is envelope["data"]
+
 
 class RabbitListenerBase:
-    def __init__(self, base, instance_name: str, handlers: Dict[str, Callable[[Any], Awaitable[None]]]):
+    def __init__(self, base, instance_name: str, handlers: Dict[str, Handler]):
         self._base = base
-        self._instance_name = instance_name
+        self._instance_name = instance_name  # queue prefix (like your .NET instance name)
         self._handlers = handlers
         self._consumers: List[aio_pika.abc.AbstractRobustQueue] = []
 
@@ -14,13 +19,15 @@ class RabbitListenerBase:
         suffix = f"-{rk_part}" if rk_part else ""
         return f"{self._instance_name}-{exchange}{suffix}"
 
-    async def start(self, declarations: List[dict]):
+    async def start(self, declarations: List[dict]) -> None:
         for d in declarations:
             exch = d["ExchangeName"]
             ttl = d.get("MessageTimeout") or None
             rks = d.get("RoutingKeys") or [""]
             qname = self._qname(exch, rks)
-            q = await self._base.declare_queue_bind(exchange=exch, queue_name=qname, routing_keys=rks, ttl_ms=ttl)
+            q = await self._base.declare_queue_bind(
+                exchange=exch, queue_name=qname, routing_keys=rks, ttl_ms=ttl
+            )
             await q.consume(self._make_consumer(d["FuncName"]))
             self._consumers.append(q)
 
@@ -31,11 +38,12 @@ class RabbitListenerBase:
             async with msg.process():
                 try:
                     envelope = json.loads(msg.body.decode("utf-8"))
-                    data = envelope.get("data", None)  # dict / list / str
+                    # Expect CloudEvent-ish envelope; we only need the 'data' field
+                    data = envelope.get("data", None)
                     if handler:
                         await handler(data)
                 except Exception:
-                    # Avoid requeue storms; log if you add a logger
+                    # Avoid requeue storms; add logging if you want
                     pass
 
         return _on_msg

rabbit_base.py

@@ -1,4 +1,4 @@
-import asyncio
+# rabbit_base.py
 from typing import Callable, Dict, List, Optional
 import aio_pika
 
@@ -12,9 +12,11 @@ class RabbitBase:
         self._conn: Optional[aio_pika.RobustConnection] = None
         self._chan: Optional[aio_pika.RobustChannel] = None
         self._exchanges: Dict[str, aio_pika.Exchange] = {}
-        self._exchange_type_resolver = exchange_type_resolver or (lambda _: settings.RABBIT_EXCHANGE_TYPE)
+        self._exchange_type_resolver = (
+            exchange_type_resolver or (lambda _: settings.RABBIT_EXCHANGE_TYPE)
+        )
 
-    async def connect(self):
+    async def connect(self) -> None:
         if self._conn and not self._conn.is_closed:
             return
         self._conn = await aio_pika.connect_robust(str(settings.AMQP_URL))
@@ -25,18 +27,28 @@ class RabbitBase:
         await self.connect()
         if name in self._exchanges:
             return self._exchanges[name]
-        ex_type = self._exchange_type_resolver(name)
-        ex = await self._chan.declare_exchange(name, getattr(aio_pika.ExchangeType, ex_type), durable=True)
+        ex_type = self._exchange_type_resolver(name)  # "fanout"|"topic"|"direct"|"headers"
+        ex = await self._chan.declare_exchange(
+            name, getattr(aio_pika.ExchangeType, ex_type), durable=True
+        )
         self._exchanges[name] = ex
         return ex
 
-    async def declare_queue_bind(self, exchange: str, queue_name: str, routing_keys: List[str], ttl_ms: Optional[int]):
+    async def declare_queue_bind(
+        self,
+        exchange: str,
+        queue_name: str,
+        routing_keys: List[str],
+        ttl_ms: Optional[int],
+    ):
         await self.connect()
         ex = await self.ensure_exchange(exchange)
-        args = {}
+        args: Dict[str, int] = {}
         if ttl_ms:
             args["x-message-ttl"] = ttl_ms
-        q = await self._chan.declare_queue(queue_name, durable=True, exclusive=False, auto_delete=True, arguments=args)
+        q = await self._chan.declare_queue(
+            queue_name, durable=True, exclusive=False, auto_delete=True, arguments=args
+        )
         for rk in routing_keys or [""]:
             await q.bind(ex, rk)
         return q

rabbit_repo.py

@@ -1,5 +1,7 @@
+# rabbit_repo.py
 import uuid
 from typing import Any, Optional
+
 import aio_pika
 
 from config import settings
@@ -20,30 +22,36 @@ class RabbitRepo(RabbitBase):
             return settings.EXCHANGE_TYPES[max(matches, key=len)]
         return settings.RABBIT_EXCHANGE_TYPE
 
-    async def publish(self, exchange: str, obj: Any, routing_key: str = ""):
+    async def publish(self, exchange: str, obj: Any, routing_key: str = "") -> None:
         ex = await self.ensure_exchange(exchange)
-        # rabbit_repo.py
+        payload = obj if not hasattr(obj, "model_dump") else obj.model_dump(by_alias=True)
         evt = CloudEvent.wrap(
             event_id=str(uuid.uuid4()),
             event_type=(obj.__class__.__name__ if obj is not None else "NullOrEmpty"),
             source=self._source,
-            data=obj if not hasattr(obj, "model_dump") else obj.model_dump(by_alias=True),
-)
+            data=payload,
+        )
         body = evt.model_dump_json(exclude_none=True).encode("utf-8")
-
         await ex.publish(aio_pika.Message(body=body), routing_key=routing_key)
 
-    async def publish_jsonz(self, exchange: str, obj: Any, routing_key: str = "", with_id: Optional[str] = None) -> str:
+    async def publish_jsonz(
+        self,
+        exchange: str,
+        obj: Any,
+        routing_key: str = "",
+        with_id: Optional[str] = None,
+    ) -> str:
         ex = await self.ensure_exchange(exchange)
-        datajson = to_json(obj if not hasattr(obj, "model_dump") else obj.model_dump(by_alias=True))
+        payload = obj if not hasattr(obj, "model_dump") else obj.model_dump(by_alias=True)
+        datajson = to_json(payload)
         datajsonZ = json_compress_str(datajson)
-        payload: Any = (datajsonZ, with_id) if with_id else datajsonZ
+        wrapped: Any = (datajsonZ, with_id) if with_id else datajsonZ
 
         evt = CloudEvent.wrap(
             event_id=str(uuid.uuid4()),
             event_type=(obj.__class__.__name__ if obj is not None else "NullOrEmpty"),
             source=self._source,
-            data=payload,
+            data=wrapped,
         )
         body = evt.model_dump_json(exclude_none=True).encode("utf-8")
         await ex.publish(aio_pika.Message(body=body), routing_key=routing_key)