# rabbit_repo.py import uuid from typing import Any, Optional import aio_pika from config import settings from models import CloudEvent from rabbit_base import RabbitBase from utils import to_json, json_compress_str import logging logger = logging.getLogger(__name__) class RabbitRepo(RabbitBase): def __init__(self, external_source: str): super().__init__(exchange_type_resolver=self._resolve_type) self._source = external_source def _resolve_type(self, exch: str) -> str: # First check for oa.* exchanges if exch.lower().startswith("oa."): return "direct" # Default for oa.* exchanges # Then check EXCHANGE_TYPES if present if hasattr(settings, 'EXCHANGE_TYPES') and settings.EXCHANGE_TYPES: matches = [k for k in settings.EXCHANGE_TYPES.keys() if exch.lower().startswith(k.lower())] if matches: return settings.EXCHANGE_TYPES[max(matches, key=len)] # Default fallback return "fanout" async def publish(self, exchange: str, obj: Any, routing_key: str = "") -> None: ex = await self.ensure_exchange(exchange) 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=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: ex = await self.ensure_exchange(exchange) payload = obj if not hasattr(obj, "model_dump") else obj.model_dump(by_alias=True) datajson = to_json(payload) datajsonZ = json_compress_str(datajson) 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=wrapped, ) body = evt.model_dump_json(exclude_none=True).encode("utf-8") await ex.publish(aio_pika.Message(body=body), routing_key=routing_key) return datajsonZ